From 0797e5def24fc8e89020f48a8aa4687f52503561 Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Tue, 6 Oct 2026 22:02:36 +0200 Subject: [PATCH] Send every log line to a syslog server as well (closes #28) With SWWAF_LOG_REMOTE_URL set (syslog+udp, syslog+tcp or syslog+tls), every line on stdout is also sent as the message of an RFC 5424 record, octet-counted over TCP and TLS, from a bounded buffer that drops its oldest line when full, so a slow or unreachable server holds up nothing. Failed connections are retried with backoff; lines sent, dropped and waiting are metrics. At a stop the lines still waiting get at most two seconds. SWWAF_LOG_REMOTE_APP_NAME defaults to SWWAF_INSTANCE_NAME; while sending, an app name RFC 5424 does not allow stops the start. Standard library only: log/syslog writes only the older format. Model: opus-5-5 --- README.md | 92 ++- internal/config/config.go | 171 ++++- internal/config/config_test.go | 204 ++++++ internal/metrics/metrics.go | 28 + internal/remotelog/remotelog.go | 310 +++++++++ internal/remotelog/remotelog_test.go | 640 ++++++++++++++++++ internal/smallwebwaf/smallwebwaf.go | 63 +- .../smallwebwaf/smallwebwaf_internal_test.go | 82 +++ internal/smallwebwaf/smallwebwaf_test.go | 171 +++++ 9 files changed, 1737 insertions(+), 24 deletions(-) create mode 100644 internal/remotelog/remotelog.go create mode 100644 internal/remotelog/remotelog_test.go create mode 100644 internal/smallwebwaf/smallwebwaf_internal_test.go diff --git a/README.md b/README.md index 1cd512d..10d0206 100644 --- a/README.md +++ b/README.md @@ -19,22 +19,23 @@ JSON state files with your edits taken in while it runs and the paths the rate limits do not count, which come next in the build order, `observe` mode and the rest of the request log's fields, which come a little later, and the metrics endpoint and the header size and the idle time as settings, which come last in -it. So are the rule files, the first part of the stage after it, with the bans -for a clear sign of attack. `smallwebwaf` passes each request to the app and the -app's answer back, unchanged, within its timeouts and size limits, works out -each client's address, bans a client that sends too many requests, not counting -those for the paths you choose, refuses a client that comes from a country you -refuse or from a network you refuse, lets the networks you choose through, -checks each request against the rule files and bans a client whose request is a -clear sign of attack, keeps its bans, each client's counters and history, and -GeoJS's answers in JSON files across restarts, takes in your edits of those -files and of the rule files while it runs, writes a JSON log line for every -request, serves Prometheus metrics to a scraper that holds the metrics token, -and in `observe` mode passes on the requests it would refuse, logging what it -would have done with them. It comes as the image the app's own image is built -on. The rest of the design comes after that, in the order of the build order in -[`SPEC.md`](SPEC.md). The survey of existing tools that led to the design is in -[`EVALUATION.md`](EVALUATION.md). +it. So are two parts of the stage after it: the rule files, the first part, with +the bans for a clear sign of attack, and remote log sending. `smallwebwaf` +passes each request to the app and the app's answer back, unchanged, within its +timeouts and size limits, works out each client's address, bans a client that +sends too many requests, not counting those for the paths you choose, refuses a +client that comes from a country you refuse or from a network you refuse, lets +the networks you choose through, checks each request against the rule files and +bans a client whose request is a clear sign of attack, keeps its bans, each +client's counters and history, and GeoJS's answers in JSON files across +restarts, takes in your edits of those files and of the rule files while it +runs, writes a JSON log line for every request, sends its log lines to a syslog +server too if you name one, serves Prometheus metrics to a scraper that holds +the metrics token, and in `observe` mode passes on the requests it would refuse, +logging what it would have done with them. It comes as the image the app's own +image is built on. The rest of the design comes after that, in the order of the +build order in [`SPEC.md`](SPEC.md). The survey of existing tools that led to +the design is in [`EVALUATION.md`](EVALUATION.md). ## Getting started @@ -171,6 +172,9 @@ in `bin/state` unless `SWWAF_STATE_DIR` is set, and the default rule file of passed to the app: a banned client stays refused, and each counts toward the client's rate limits. None of them reaches the app. - Writes a line in the request log for each request (see "Request log" below). +- Sends every line it writes on stdout to a syslog server as well, while + `SWWAF_LOG_REMOTE_URL` names one (see "Sending the log to a syslog server" + below). ## Settings @@ -288,6 +292,23 @@ it, and the effective settings are logged at start. rule files. A directory that does not exist stops the start. - `SWWAF_RULES_ENABLED` (default `true`): `false` reads no rule file, and checks no request against one. +- `SWWAF_LOG_REMOTE_URL` (default unset): a syslog server that every line on + stdout is also sent to, as `syslog+udp://`, `syslog+tcp://` or `syslog+tls://` + with a host and a port, such as `syslog+tls://logs.example:6514`. Unset or + empty, nothing is sent. +- `SWWAF_LOG_REMOTE_TLS_CA_FILE` (default unset): a file of PEM certificates, + which the certificate of a `syslog+tls` server must chain to instead of the + host's own. A file that cannot be read or holds no certificate stops the + start. +- `SWWAF_LOG_REMOTE_BUFFER` (default `10000`): the most lines held while they + wait to be sent. +- `SWWAF_LOG_REMOTE_FACILITY` (default `local0`): the syslog facility the lines + are sent with: `kern`, `user`, `mail`, `daemon`, `auth`, `syslog`, `lpr`, + `news`, `uucp`, `cron`, `authpriv`, `ftp`, or `local0` to `local7`. +- `SWWAF_LOG_REMOTE_APP_NAME` (default `SWWAF_INSTANCE_NAME`): the app name the + lines are sent with, 1 to 48 printable ASCII characters without a space. While + `SWWAF_LOG_REMOTE_URL` is set, an `SWWAF_INSTANCE_NAME` that is not such a + name stops the start too, unless this setting gives one that is. Durations are in Go's syntax, with `d` for days (`90s`, `15m`, `7d`). Sizes are bytes, with an optional `K`, `M` or `G`, which are powers of 1024 (`1K` is 1024 @@ -297,8 +318,8 @@ ISO 3166-1 assigns today, and `xk` for Kosovo, in either case (`de` and `DE` are the same); any other code, such as `nk` (North Korea is `kp`) or the withdrawn `su`, stops the start, and so does a code on both country lists. `off` switches a timeout, a size limit or a rate limit off; -`SWWAF_CLIENT_REQUEST_HEADER_MAX_BYTES`, the ban settings, the state settings -and `SWWAF_METRICS_TOP_N` cannot be off. +`SWWAF_CLIENT_REQUEST_HEADER_MAX_BYTES`, the ban settings, the state settings, +`SWWAF_METRICS_TOP_N` and `SWWAF_LOG_REMOTE_BUFFER` cannot be off. Several limits are fixed rather than settings. At most 20,000 clients are kept, with their counters and history, and an IPv6 client is counted by its /64. A new @@ -404,6 +425,32 @@ which it answers `431`, headers slower than `SWWAF_CLIENT_REQUEST_TIMEOUT`, whose connection it closes without an answer, and requests it cannot read at all, which it answers itself, mostly with `400`. +### Sending the log to a syslog server + +While `SWWAF_LOG_REMOTE_URL` is set, every line `smallwebwaf` writes on stdout, +request lines and its own, is also sent to that syslog server, as the message of +an RFC 5424 record: one record to a datagram over UDP, and over TCP and TLS each +record after its length in bytes and a space. A record gives the facility +`SWWAF_LOG_REMOTE_FACILITY` names, the severity informational, the time the line +was written, in the same form as a request line's `time`, the host's name, and +the app name `SWWAF_LOG_REMOTE_APP_NAME` gives. stdout is unchanged. + +The lines wait in a buffer of `SWWAF_LOG_REMOTE_BUFFER` lines and are sent from +there, so a server that is slow or cannot be reached never holds up a request or +stdout. When the buffer is full, its oldest line is dropped to make room. A line +whose sending fails is dropped too, and the connection closed. That failure, +like a failed attempt to connect, is logged and followed by the next attempt to +connect a second later, twice as long after each further failure up to a minute, +and a second again after a connection that stayed up for a minute before it +failed. A line too long for one UDP datagram is dropped alone, with no wait and +nothing logged. UDP gives no sign of what arrives, and over TCP and TLS a line +sent on a connection the server has just closed can be lost before a failure +shows; such a loss is not counted. + +As `smallwebwaf` stops, it sends the lines still waiting, on the connection open +or a new one, for at most two seconds, and gives up the rest; stdout has carried +them. + ## State files `smallwebwaf` keeps its state in memory and a copy of it in three JSON files in @@ -604,6 +651,11 @@ other request. No metric carries a client's address. `smallwebwaf_state_file_edits_taken_in_total`: your edits taken in, and `smallwebwaf_state_file_edits_set_aside_total`: those renamed to `.bad` because they would stop the start. +- While `SWWAF_LOG_REMOTE_URL` is set, + `smallwebwaf_remote_log_lines_sent_total`: the lines sent to it; + `smallwebwaf_remote_log_lines_dropped_total`: those dropped, from a full + buffer or because their sending failed; and + `smallwebwaf_remote_log_buffer_depth`: those waiting in the buffer. - Go's own `go_` metrics and the process's `process_` metrics. The requests Go's HTTP server ends before `smallwebwaf` sees them (see "Request @@ -913,6 +965,10 @@ addresses are never sent to GeoJS. one while running, and writes them when they are due and at the stop. - `internal/requestlog`: the lines on stdout: the request log line and the process's own messages. +- `internal/remotelog`: sends the lines on stdout to `SWWAF_LOG_REMOTE_URL`, + each as a syslog record, from a buffer of its own. It is written with the + standard library alone, whose `log/syslog` writes only the older syslog + format. - `Dockerfile`: the lint and test phases, then the image, whose last stage installs Ubuntu's packages, nixpkgs, `runsvinit` and `smallwebwaf`, with `share/smallwebwaf.run` as runit's `run` script for `smallwebwaf` and diff --git a/internal/config/config.go b/internal/config/config.go index 941903e..0f0ddc0 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,6 +4,7 @@ package config import ( + "crypto/x509" "errors" "fmt" "log/slog" @@ -19,6 +20,8 @@ import ( "strings" "time" "unicode/utf8" + + "sneak.berlin/go/smallwebwaf/internal/remotelog" ) // Config is smallwebwaf's settings. A timeout, size or rate limit of zero @@ -135,6 +138,21 @@ type Config struct { // unless RulesEnabled is false (SWWAF_RULES_ENABLED). RulesDir string RulesEnabled bool + // LogRemoteURL is where every line on stdout is also sent + // (SWWAF_LOG_REMOTE_URL), nil while it is unset and nothing is sent. + // LogRemoteTLSCAs are the certificates a syslog+tls endpoint's + // certificate must chain to (SWWAF_LOG_REMOTE_TLS_CA_FILE), nil while + // it is unset and the host's own are used. LogRemoteBuffer is the most + // lines held while they wait to be sent (SWWAF_LOG_REMOTE_BUFFER). + // LogRemoteFacility is the number of the syslog facility + // (SWWAF_LOG_REMOTE_FACILITY), and LogRemoteAppName the APP-NAME + // (SWWAF_LOG_REMOTE_APP_NAME, by default InstanceName), of the records + // the lines are sent in. + LogRemoteURL *url.URL + LogRemoteTLSCAs *x509.CertPool + LogRemoteBuffer int + LogRemoteFacility int + LogRemoteAppName string // settings are the values read, as given or by default, for the // log line at start. @@ -195,7 +213,14 @@ var ( errNotMode = errors.New("is not enforce or observe") errNotPathPrefix = errors.New( "is not a path prefix starting with /, such as /assets/") - errNotBoolean = errors.New("is not true or false") + errNotBoolean = errors.New("is not true or false") + errNotLogRemoteURL = errors.New( + "is not syslog+udp, syslog+tcp or syslog+tls with a host and a port, " + + "and nothing more, such as syslog+tls://logs.example:6514") + errNoCertificate = errors.New("holds no PEM certificate") + errNotFacility = errors.New("is not a syslog facility such as local0 or daemon") + errNotAppName = errors.New( + "is not 1 to 48 printable ASCII characters without a space, such as gitea") ) // FromEnvironment reads the settings with lookupEnv, normally @@ -241,12 +266,19 @@ func FromEnvironment(lookupEnv func(string) (string, bool)) (*Config, error) { StateCounterInterval: env.durationNotOff("SWWAF_STATE_COUNTER_INTERVAL", "15m"), LogRequestHeaders: env.headerNames("SWWAF_LOG_REQUEST_HEADERS", "accept,accept-language,accept-encoding,content-type,origin,range"), - MetricsToken: env.token("SWWAF_METRICS_TOKEN"), - MetricsTopN: env.numberNotOff("SWWAF_METRICS_TOP_N", "50"), - RulesDir: env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"), - RulesEnabled: env.boolean("SWWAF_RULES_ENABLED", "true"), + MetricsToken: env.token("SWWAF_METRICS_TOKEN"), + MetricsTopN: env.numberNotOff("SWWAF_METRICS_TOP_N", "50"), + RulesDir: env.value("SWWAF_RULES_DIR", "/etc/smallwebwaf/rules.d"), + RulesEnabled: env.boolean("SWWAF_RULES_ENABLED", "true"), + LogRemoteURL: env.logRemoteURL("SWWAF_LOG_REMOTE_URL"), + LogRemoteTLSCAs: env.certificates("SWWAF_LOG_REMOTE_TLS_CA_FILE"), + LogRemoteBuffer: env.numberNotOff("SWWAF_LOG_REMOTE_BUFFER", "10000"), + LogRemoteFacility: env.facility("SWWAF_LOG_REMOTE_FACILITY", "local0"), } + cfg.LogRemoteAppName = env.appName("SWWAF_LOG_REMOTE_APP_NAME", + cfg.InstanceName, cfg.LogRemoteURL != nil) + for _, country := range cfg.ExclusivelyAllowedCountries { if slices.Contains(cfg.DeniedCountries, country) { env.check("SWWAF_EXCLUSIVELY_ALLOWED_COUNTRIES", @@ -468,6 +500,74 @@ func (e *environment) token(name string) string { return value } +// logRemoteURL reads the setting that is where every log line is also +// sent. Unset or empty, it is nil, and nothing is sent. +func (e *environment) logRemoteURL(name string) *url.URL { + value := e.value(name, "") + if value == "" { + return nil + } + + remote, err := parseLogRemoteURL(value) + e.check(name, err) + + return remote +} + +// certificates reads a setting that is the path of a file of PEM +// certificates. Unset or empty, it is nil. +func (e *environment) certificates(name string) *x509.CertPool { + path := e.value(name, "") + if path == "" { + return nil + } + + pem, err := os.ReadFile(path) //nolint:gosec // a file the admin names + if err != nil { + e.check(name, fmt.Errorf("cannot be read: %w", err)) + + return nil + } + + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pem) { + e.check(name, fmt.Errorf("%q %w", path, errNoCertificate)) + + return nil + } + + return pool +} + +// facility reads a setting that is a syslog facility, and returns its +// number. +func (e *environment) facility(name, defaultValue string) int { + number, err := parseFacility(e.value(name, defaultValue)) + e.check(name, err) + + return number +} + +// appName reads the setting that is the APP-NAME of the records the log +// lines are sent in, by default the instance name. Its value is checked +// when it is set, and, while lines are sent, when it is the instance name. +func (e *environment) appName(name, instanceName string, sending bool) string { + _, set := e.lookupEnv(name) + + value := e.value(name, instanceName) + + switch { + case isAppName(value): + case set: + e.check(name, fmt.Errorf("%q %w", value, errNotAppName)) + case sending: + e.check(name, fmt.Errorf("is unset, and SWWAF_INSTANCE_NAME %q, its default, %w", + value, errNotAppName)) + } + + return value +} + // parseDuration reads a duration in Go's syntax, such as 90s or 15m, a // whole number of days such as 7d, or off. func parseDuration(value string) (time.Duration, error) { @@ -841,3 +941,64 @@ func parseUpstreamURL(value string) (*url.URL, error) { return upstream, nil } + +// parseLogRemoteURL reads where every log line is also sent: +// syslog+udp, syslog+tcp or syslog+tls, a host and a port from 1 to +// 65535, and nothing else. +func parseLogRemoteURL(value string) (*url.URL, error) { + remote, err := url.Parse(value) + if err != nil { + return nil, fmt.Errorf("%q %w", value, errNotLogRemoteURL) + } + + schemes := []string{remotelog.SchemeUDP, remotelog.SchemeTCP, remotelog.SchemeTLS} + port, err := strconv.ParseUint(remote.Port(), 10, 16) + + onlySchemeHostAndPort := slices.Contains(schemes, remote.Scheme) && + remote.Hostname() != "" && err == nil && port != 0 && + remote.User == nil && remote.Opaque == "" && + (remote.Path == "" || remote.Path == "/") && + remote.RawQuery == "" && remote.Fragment == "" + if !onlySchemeHostAndPort { + return nil, fmt.Errorf("%q %w", value, errNotLogRemoteURL) + } + + return remote, nil +} + +// parseFacility reads the name of a syslog facility, and returns its +// number, as RFC 5424 numbers them. +func parseFacility(value string) (int, error) { + //nolint:mnd // the facilities' numbers in RFC 5424 + number, known := map[string]int{ + "kern": 0, "user": 1, "mail": 2, "daemon": 3, "auth": 4, "syslog": 5, + "lpr": 6, "news": 7, "uucp": 8, "cron": 9, "authpriv": 10, "ftp": 11, + "local0": 16, "local1": 17, "local2": 18, "local3": 19, + "local4": 20, "local5": 21, "local6": 22, "local7": 23, + }[value] + if !known { + return 0, fmt.Errorf("%q %w", value, errNotFacility) + } + + return number, nil +} + +// appNameMaxLength is the most characters RFC 5424 allows in an +// APP-NAME. +const appNameMaxLength = 48 + +// isAppName reports whether value can be an APP-NAME: 1 to +// appNameMaxLength printable ASCII characters, none of them a space. +func isAppName(value string) bool { + if value == "" || len(value) > appNameMaxLength { + return false + } + + for _, char := range []byte(value) { + if char < '!' || char > '~' { + return false + } + } + + return true +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a9c326a..30a97e1 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -2,11 +2,13 @@ package config_test import ( "bytes" + "crypto/x509" "encoding/json" "log/slog" "maps" "net/netip" "os" + "path/filepath" "slices" "strings" "testing" @@ -54,15 +56,41 @@ const ( logRequestHeaders = "SWWAF_LOG_REQUEST_HEADERS" rulesDir = "SWWAF_RULES_DIR" rulesEnabled = "SWWAF_RULES_ENABLED" + logRemoteURL = "SWWAF_LOG_REMOTE_URL" + logRemoteTLSCAFile = "SWWAF_LOG_REMOTE_TLS_CA_FILE" + logRemoteBuffer = "SWWAF_LOG_REMOTE_BUFFER" + logRemoteFacility = "SWWAF_LOG_REMOTE_FACILITY" + logRemoteAppName = "SWWAF_LOG_REMOTE_APP_NAME" ) // defaultLogRequestHeaders is the default of SWWAF_LOG_REQUEST_HEADERS. const defaultLogRequestHeaders = "accept,accept-language,accept-encoding," + "content-type,origin,range" +// testCA is a CA certificate, of which only that it reads matters here. +const testCA = `-----BEGIN CERTIFICATE----- +MIIBkzCCATmgAwIBAgIUeySaE27dnr6A2HijrMB13gTLUKIwCgYIKoZIzj0EAwIw +HjEcMBoGA1UEAwwTc21hbGx3ZWJ3YWYgdGVzdCBDQTAgFw0yNjEwMDYxNDI2MTNa +GA8yMTI2MDkxMjE0MjYxM1owHjEcMBoGA1UEAwwTc21hbGx3ZWJ3YWYgdGVzdCBD +QTBZMBMGByqGSM49AgEGCCqGSM49AwEHA0IABKDEhcWKKhet2KgSdME+iEPxyEyn +2sd9IdElbt8DM2SfCdB2JsXo0C07UNZaywMPMfn/n8LNI/PKwu+N2uX7gfSjUzBR +MB0GA1UdDgQWBBRZo3BPLv0KbV4drw6JI1JIUKmoRzAfBgNVHSMEGDAWgBRZo3BP +Lv0KbV4drw6JI1JIUKmoRzAPBgNVHRMBAf8EBTADAQH/MAoGCCqGSM49BAMCA0gA +MEUCICC5k+76UpWoSwVbZA+atu5WcALEOJGqwOUWua3zemhcAiEA8Hfdxgwp0z2v +rlG9y/jrJb6ORy3kTLWo2EA0BA67vuI= +-----END CERTIFICATE----- +` + // token is a token of 32 characters, the shortest allowed. const token = "0123456789abcdef0123456789abcdef" +// instance is an SWWAF_INSTANCE_NAME that is a valid app name too, and +// remoteURL an SWWAF_LOG_REMOTE_URL, for the tests that send the lines. +const ( + instance = "fsn1app1/gitea" + remoteURL = "syslog+udp://192.0.2.1:514" +) + // off switches a timeout, a size limit or a rate limit off. const off = "off" @@ -273,6 +301,177 @@ func TestInstanceNameAndLoggedHeadersAsSet(t *testing.T) { } } +func TestRemoteLogSettingsDefaults(t *testing.T) { + t.Parallel() + + cfg := fromEnvironment(t, environment{instanceName: instance}) + + if cfg.LogRemoteURL != nil || cfg.LogRemoteTLSCAs != nil || + cfg.LogRemoteBuffer != 10000 || cfg.LogRemoteFacility != 16 || + cfg.LogRemoteAppName != instance { + t.Errorf("remote log settings %v, %v, %d, %d and %q, want no URL, no "+ + "certificates, 10000, 16 and %s's %s", cfg.LogRemoteURL, + cfg.LogRemoteTLSCAs, cfg.LogRemoteBuffer, cfg.LogRemoteFacility, + cfg.LogRemoteAppName, instanceName, instance) + } +} + +func TestRemoteLogSettingsAsSet(t *testing.T) { + t.Parallel() + + caFile := filepath.Join(t.TempDir(), "ca.pem") + + err := os.WriteFile(caFile, []byte(testCA), 0o600) + if err != nil { + t.Fatalf("write %s: %v", caFile, err) + } + + cfg := fromEnvironment(t, environment{ + logRemoteURL: "syslog+tls://logs.example:6514", + logRemoteTLSCAFile: caFile, + logRemoteBuffer: "500", + logRemoteFacility: "daemon", + logRemoteAppName: instance, + }) + + roots := x509.NewCertPool() + roots.AppendCertsFromPEM([]byte(testCA)) + + if cfg.LogRemoteURL.String() != "syslog+tls://logs.example:6514" || + !roots.Equal(cfg.LogRemoteTLSCAs) || cfg.LogRemoteBuffer != 500 || + cfg.LogRemoteFacility != 3 || cfg.LogRemoteAppName != instance { + t.Errorf("remote log settings %v, %v, %d, %d and %q", cfg.LogRemoteURL, + cfg.LogRemoteTLSCAs, cfg.LogRemoteBuffer, cfg.LogRemoteFacility, + cfg.LogRemoteAppName) + } +} + +func TestRemoteLogURLForms(t *testing.T) { + t.Parallel() + + for _, value := range []string{ + "syslog+udp://192.0.2.1:514", + "syslog+tcp://[2001:db8::1]:514", + "syslog+tls://logs.example:6514/", + } { + cfg := fromEnvironment(t, environment{logRemoteURL: value}) + if cfg.LogRemoteURL.String() != value { + t.Errorf("%s read as %v", value, cfg.LogRemoteURL) + } + } + + cfg := fromEnvironment(t, environment{logRemoteURL: ""}) + if cfg.LogRemoteURL != nil { + t.Errorf("set but empty, %s read as %v", logRemoteURL, cfg.LogRemoteURL) + } +} + +func TestRemoteLogFacilitiesByNumber(t *testing.T) { + t.Parallel() + + for name, number := range map[string]int{ + "kern": 0, "user": 1, "auth": 4, "authpriv": 10, "ftp": 11, + "local0": 16, "local5": 21, "local7": 23, + } { + cfg := fromEnvironment(t, environment{logRemoteFacility: name}) + if cfg.LogRemoteFacility != number { + t.Errorf("%s read as %d, want %d", name, cfg.LogRemoteFacility, number) + } + } +} + +func TestInvalidRemoteLogSettingStopsTheStart(t *testing.T) { + t.Parallel() + + for _, tc := range []struct{ name, value string }{ + {logRemoteURL, "logs.example:514"}, + {logRemoteURL, "syslog://logs.example:514"}, + {logRemoteURL, "http://logs.example:514"}, + {logRemoteURL, "syslog+udp://logs.example"}, + {logRemoteURL, "syslog+tcp://:514"}, + {logRemoteURL, "syslog+tcp://logs.example:0"}, + {logRemoteURL, "syslog+tls://logs.example:65536"}, + {logRemoteURL, "syslog+tls://user@logs.example:6514"}, + {logRemoteURL, "syslog+tcp://logs.example:514/app"}, + {logRemoteURL, "syslog+tcp://logs.example:514?tls=1"}, + {logRemoteTLSCAFile, "/nonexistent/ca.pem"}, + {logRemoteBuffer, off}, {logRemoteBuffer, "0"}, {logRemoteBuffer, "10K"}, + {logRemoteFacility, "local8"}, {logRemoteFacility, "LOCAL0"}, + {logRemoteFacility, "16"}, {logRemoteFacility, ""}, + {logRemoteAppName, ""}, {logRemoteAppName, "my app"}, + {logRemoteAppName, "gitéa"}, {logRemoteAppName, strings.Repeat("a", 49)}, + } { + _, err := config.FromEnvironment(environment{tc.name: tc.value}.lookupEnv) + if err == nil || !strings.HasPrefix(err.Error(), tc.name+": ") { + t.Errorf("%s=%q: error %v, want one naming it", tc.name, tc.value, err) + } + } +} + +func TestRemoteLogCAFileWithoutCertificateStopsTheStart(t *testing.T) { + t.Parallel() + + caFile := filepath.Join(t.TempDir(), "ca.pem") + + err := os.WriteFile(caFile, []byte("not a certificate\n"), 0o600) + if err != nil { + t.Fatalf("write %s: %v", caFile, err) + } + + _, err = config.FromEnvironment(environment{logRemoteTLSCAFile: caFile}.lookupEnv) + + want := logRemoteTLSCAFile + `: "` + caFile + `" holds no PEM certificate` + if err == nil || err.Error() != want { + t.Errorf("error %v, want %s", err, want) + } +} + +func TestInstanceNameNotAnAppNameStopsTheStartOnlyWhileSending(t *testing.T) { + t.Parallel() + + const spaced = "fsn1 app1" + + sending := environment{logRemoteURL: remoteURL, instanceName: spaced} + + _, err := config.FromEnvironment(sending.lookupEnv) + + want := logRemoteAppName + `: is unset, and ` + instanceName + + ` "fsn1 app1", its default, is not 1 to 48 printable ASCII characters ` + + `without a space, such as gitea` + if err == nil || err.Error() != want { + t.Errorf("error %v, want %s", err, want) + } + + cfg := fromEnvironment(t, environment{instanceName: spaced}) + if cfg.LogRemoteAppName != spaced { + t.Errorf("not sending, %s is %q", logRemoteAppName, cfg.LogRemoteAppName) + } + + sending[logRemoteAppName] = instance + + cfg = fromEnvironment(t, sending) + if cfg.LogRemoteAppName != instance { + t.Errorf("set to %s, %s is %q", instance, logRemoteAppName, + cfg.LogRemoteAppName) + } +} + +func TestAppNameSetStopsTheStartWhileSending(t *testing.T) { + t.Parallel() + + _, err := config.FromEnvironment(environment{ + logRemoteURL: remoteURL, + instanceName: instance, + logRemoteAppName: "my app", + }.lookupEnv) + + want := logRemoteAppName + `: "my app" is not 1 to 48 printable ASCII ` + + `characters without a space, such as gitea` + if err == nil || err.Error() != want { + t.Errorf("error %v, want %s", err, want) + } +} + func TestCodeOnBothCountryListsStopsTheStart(t *testing.T) { t.Parallel() @@ -581,6 +780,11 @@ func TestLogsEachSettingWithItsValue(t *testing.T) { logRequestHeaders: defaultLogRequestHeaders, rulesDir: "/etc/smallwebwaf/rules.d", rulesEnabled: "true", + logRemoteURL: "", + logRemoteTLSCAFile: "", + logRemoteBuffer: "10000", + logRemoteFacility: "local0", + logRemoteAppName: hostname, } if !maps.Equal(line.Settings, want) { t.Errorf("logged settings\n%v\nwant\n%v", line.Settings, want) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 52f969a..96acb0a 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -13,6 +13,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "sneak.berlin/go/smallwebwaf/internal/bans" "sneak.berlin/go/smallwebwaf/internal/ratelimit" + "sneak.berlin/go/smallwebwaf/internal/remotelog" "sneak.berlin/go/smallwebwaf/internal/requestlog" "sneak.berlin/go/smallwebwaf/internal/rules" ) @@ -197,6 +198,33 @@ func (m *Metrics) AddRules(ruleFiles *rules.Files) { })) } +// AddRemoteLog adds the metrics of sending the log lines to +// SWWAF_LOG_REMOTE_URL, read from remote as the metrics are asked for: the +// lines sent, those dropped, and those waiting in the buffer. +func (m *Metrics) AddRemoteLog(remote *remotelog.Sender) { + m.registry.MustRegister( + prometheus.NewCounterFunc(prometheus.CounterOpts{ + Name: "smallwebwaf_remote_log_lines_sent_total", + Help: "Log lines sent to SWWAF_LOG_REMOTE_URL.", + }, func() float64 { + return float64(remote.Sent()) + }), + prometheus.NewCounterFunc(prometheus.CounterOpts{ + Name: "smallwebwaf_remote_log_lines_dropped_total", + Help: "Log lines dropped: the oldest in a full buffer, and those " + + "whose sending failed.", + }, func() float64 { + return float64(remote.Dropped()) + }), + prometheus.NewGaugeFunc(prometheus.GaugeOpts{ + Name: "smallwebwaf_remote_log_buffer_depth", + Help: "Log lines in the buffer, waiting to be sent.", + }, func() float64 { + return float64(remote.Depth()) + }), + ) +} + // ServeHTTP answers with the metrics in the Prometheus text format. func (m *Metrics) ServeHTTP(w http.ResponseWriter, r *http.Request) { m.handler.ServeHTTP(w, r) diff --git a/internal/remotelog/remotelog.go b/internal/remotelog/remotelog.go new file mode 100644 index 0000000..d40239b --- /dev/null +++ b/internal/remotelog/remotelog.go @@ -0,0 +1,310 @@ +// Package remotelog sends the lines smallwebwaf writes on stdout to the +// remote log endpoint, SWWAF_LOG_REMOTE_URL, as the "Request log" section +// of SPEC.md describes: each line as the message of an RFC 5424 syslog +// record, over UDP, TCP or TLS. Lines wait in a bounded buffer, so a slow +// or unreachable endpoint never holds up a request or stdout. +package remotelog + +import ( + "bytes" + "context" + "crypto/tls" + "crypto/x509" + "errors" + "fmt" + "log/slog" + "net" + "net/url" + "os" + "strconv" + "sync/atomic" + "syscall" + "time" + + "sneak.berlin/go/smallwebwaf/internal/requestlog" +) + +// The forms of SWWAF_LOG_REMOTE_URL, by its scheme. +const ( + SchemeUDP = "syslog+udp" + SchemeTCP = "syslog+tcp" + SchemeTLS = "syslog+tls" +) + +// A record's priority is the number of its facility times the number of +// severities there are, plus the number of its severity. Every record's +// severity is informational. +const ( + severities = 8 + informational = 6 +) + +const ( + // dialTimeout bounds connecting to the endpoint, the TLS handshake + // included. + dialTimeout = 10 * time.Second + // After a failed attempt to connect, or a connection on which a record + // fails, the next attempt to connect is made a second later, and + // retryDelayFactor times as long after each further failure in a row, + // up to a minute. A connection that fails after it has stayed up for + // resetRetryDelayAfter ends the row. + firstRetryDelay = time.Second + retryDelayFactor = 2 + maxRetryDelay = time.Minute + resetRetryDelayAfter = time.Minute +) + +// Params are what New needs. +type Params struct { + // URL is the endpoint (SWWAF_LOG_REMOTE_URL): SchemeUDP, SchemeTCP or + // SchemeTLS, a host and a port. + URL *url.URL + // RootCAs are the certificates a SchemeTLS endpoint's certificate + // must chain to (SWWAF_LOG_REMOTE_TLS_CA_FILE), nil for the host's. + RootCAs *x509.CertPool + // Buffer is the most lines held while they wait to be sent + // (SWWAF_LOG_REMOTE_BUFFER). + Buffer int + // Facility is the number of the records' syslog facility + // (SWWAF_LOG_REMOTE_FACILITY), and AppName their APP-NAME + // (SWWAF_LOG_REMOTE_APP_NAME). + Facility int + AppName string +} + +// Sender sends lines to the endpoint. Write puts them in its buffer, and +// Run sends them from there. +type Sender struct { + url *url.URL + tlsConfig *tls.Config + // beforeTime and afterTime are the parts of every record's header + // before and after its time, as RFC 5424 lays the header out. + beforeTime string + afterTime string + // records is the buffer: each line's record, framed to be sent. + records chan []byte + sent atomic.Int64 + dropped atomic.Int64 +} + +// New returns a Sender for the endpoint params.URL. +func New(params Params) *Sender { + hostname, err := os.Hostname() + if err != nil || hostname == "" { + hostname = "-" // RFC 5424's value for a field that has none + } + + priority := params.Facility*severities + informational + + return &Sender{ + url: params.URL, + tlsConfig: &tls.Config{ + RootCAs: params.RootCAs, + MinVersion: tls.VersionTLS12, + }, + // The 1 is the version of the format. The process id, the message + // id and the structured data have no value. + beforeTime: "<" + strconv.Itoa(priority) + ">1 ", + afterTime: " " + hostname + " " + params.AppName + " - - - ", + records: make(chan []byte, params.Buffer), + } +} + +// Write puts each line in p in the buffer, as the message of a record of +// its own, and never waits: when the buffer is full, the oldest record in +// it is dropped to make room. It is safe for concurrent use. +func (s *Sender) Write(p []byte) (int, error) { + at := requestlog.FormatTime(time.Now()) + + for line := range bytes.Lines(p) { + line = bytes.TrimSuffix(line, []byte("\n")) + if len(line) > 0 { + s.put(s.record(at, line)) + } + } + + return len(p), nil +} + +// Sent is how many records have been sent. +func (s *Sender) Sent() int64 { + return s.sent.Load() +} + +// Dropped is how many records were dropped: the oldest in a full buffer, +// and those whose sending failed. +func (s *Sender) Dropped() int64 { + return s.dropped.Load() +} + +// Depth is how many records are in the buffer. +func (s *Sender) Depth() int { + return len(s.records) +} + +// Run connects to the endpoint and sends each record as it comes into the +// buffer, until ctx is done. Then it sends the records still in the buffer, +// on the connection open at that time or, if there is none, on a new one, +// until none is left or one fails, and returns. How long it may take over +// that is for the caller to bound. +// +// A connection on which a record fails is closed and the record dropped. +// That failure, like a failed attempt to connect, is logged to processLog +// and followed by the next attempt after firstRetryDelay, retryDelayFactor +// times as long after each further failure in a row up to maxRetryDelay, +// and firstRetryDelay again after a connection that stayed up for +// resetRetryDelayAfter. Meanwhile the records wait in the buffer. +func (s *Sender) Run(ctx context.Context, processLog *slog.Logger) { + conn := s.send(ctx, processLog) + if conn == nil && len(s.records) > 0 { + conn, _ = s.dial(context.WithoutCancel(ctx)) + } + + if conn == nil { + return + } + + defer func() { + _ = conn.Close() + }() + + for { + select { + case record := <-s.records: + if s.write(conn, record) != nil { + return + } + default: + return + } + } +} + +// record returns line as an RFC 5424 record made at the time at, framed +// for the endpoint: on its own over UDP, since each datagram holds one, +// and over TCP and TLS after its length in bytes and a space, the +// octet-counted framing of RFC 6587 and RFC 5425. +func (s *Sender) record(at string, line []byte) []byte { + record := make([]byte, 0, len(s.beforeTime)+len(at)+len(s.afterTime)+len(line)) + record = append(record, s.beforeTime...) + record = append(record, at...) + record = append(record, s.afterTime...) + record = append(record, line...) + + if s.url.Scheme == SchemeUDP { + return record + } + + return append([]byte(strconv.Itoa(len(record))+" "), record...) +} + +// put adds record to the buffer, first dropping the oldest record in it +// while it is full. +func (s *Sender) put(record []byte) { + for { + select { + case s.records <- record: + return + default: + } + + select { + case <-s.records: + s.dropped.Add(1) + default: + } + } +} + +// send connects to the endpoint and sends each record as it comes into +// the buffer, until ctx is done, and returns the connection then open, or +// nil. +func (s *Sender) send(ctx context.Context, processLog *slog.Logger) net.Conn { + delay := firstRetryDelay + + for { + conn, err := s.dial(ctx) + if ctx.Err() != nil { + return conn + } + + if err == nil { + connected := time.Now() + + err = s.sendOn(ctx, conn) + if err == nil { + return conn + } + + _ = conn.Close() + + if time.Since(connected) >= resetRetryDelayAfter { + delay = firstRetryDelay + } + } + + processLog.Warn("sending to SWWAF_LOG_REMOTE_URL failed", + "error", err.Error(), "connecting_again_in", delay.String()) + + select { + case <-time.After(delay): + case <-ctx.Done(): + return nil + } + + delay = min(retryDelayFactor*delay, maxRetryDelay) + } +} + +// sendOn sends each record on conn as it comes into the buffer, until one +// fails, whose error it returns, or ctx is done. +func (s *Sender) sendOn(ctx context.Context, conn net.Conn) error { + for { + select { + case record := <-s.records: + err := s.write(conn, record) + if err != nil { + return err + } + case <-ctx.Done(): + return nil + } + } +} + +// write sends record on conn, and counts it as sent or, if that fails, +// as dropped. A record too long for one UDP datagram is dropped without +// an error, since the connection has not failed: a long request must not +// hold up the lines after it. +func (s *Sender) write(conn net.Conn, record []byte) error { + _, err := conn.Write(record) + if err != nil { + s.dropped.Add(1) + + if errors.Is(err, syscall.EMSGSIZE) { + return nil + } + + return fmt.Errorf("send a record: %w", err) + } + + s.sent.Add(1) + + return nil +} + +// dial connects to the endpoint. +func (s *Sender) dial(ctx context.Context) (net.Conn, error) { + dialer := &net.Dialer{Timeout: dialTimeout} + + switch s.url.Scheme { + case SchemeUDP: + return dialer.DialContext(ctx, "udp", s.url.Host) + case SchemeTLS: + tlsDialer := &tls.Dialer{NetDialer: dialer, Config: s.tlsConfig} + + return tlsDialer.DialContext(ctx, "tcp", s.url.Host) + default: + return dialer.DialContext(ctx, "tcp", s.url.Host) + } +} diff --git a/internal/remotelog/remotelog_test.go b/internal/remotelog/remotelog_test.go new file mode 100644 index 0000000..10b02fe --- /dev/null +++ b/internal/remotelog/remotelog_test.go @@ -0,0 +1,640 @@ +package remotelog_test + +import ( + "bufio" + "bytes" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/json" + "fmt" + "io" + "log/slog" + "math/big" + "net" + "net/url" + "os" + "slices" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "sneak.berlin/go/smallwebwaf/internal/remotelog" +) + +// The tests run in a synctest bubble, where the time package runs on a +// clock of the test's own, which starts at 2000-01-01T00:00:00Z: a wait +// lasts exactly as long as it should, however slowly the test process +// runs, and synctest.Wait returns once the sender has done all it can +// before time passes. The endpoint is a listener on the loopback address. +// A test reads from it only once the records are on their way, and checks +// the sender's counts first, since a goroutine of the bubble that waits on +// the network keeps that clock from moving on. For the same reason the +// endpoint that refuses connections, a tlsEndpoint, runs outside the +// bubble: a sender connecting over TLS waits on the endpoint's answer. + +const ( + // started is the time a record made as a test starts gives. + started = "2000-01-01T00:00:00.000Z" + appName = "fsn1app1/gitea" + // local0 is the number of the default facility, and local0Info the + // priority of its records. + local0 = 16 + local0Info = "<134>" + // loopback is where the endpoints listen. + loopback = "127.0.0.1:0" +) + +func TestRecordsOverUDPGoOnePerDatagram(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + endpoint, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", loopback) + if err != nil { + t.Fatalf("listen: %v", err) + } + + t.Cleanup(func() { _ = endpoint.Close() }) + + sender, _, _ := run(t, params(remotelog.SchemeUDP, endpoint.LocalAddr())) + + _, _ = sender.Write([]byte(`{"type":"request"}` + "\n" + `{"type":"process"}` + "\n")) + + synctest.Wait() + wantCounts(t, sender, 2, 0, 0) + + for _, line := range []string{`{"type":"request"}`, `{"type":"process"}`} { + datagram := make([]byte, 1024) + + n, _, err := endpoint.ReadFrom(datagram) + if err != nil { + t.Fatalf("read: %v", err) + } + + want := record(t, local0Info, appName, line) + if string(datagram[:n]) != want { + t.Errorf("datagram %q, want %q", datagram[:n], want) + } + } + }) +} + +func TestRecordsOverTCPAreOctetCountedWithTheirFacility(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + endpoint := listen(t) + endpointParams := params(remotelog.SchemeTCP, endpoint.Addr()) + endpointParams.Facility = 19 // local3 + endpointParams.AppName = "gitea" + sender, _, _ := run(t, endpointParams) + + _, _ = sender.Write([]byte("first\nsecond\n")) + + synctest.Wait() + wantCounts(t, sender, 2, 0, 0) + + frames := bufio.NewReader(accept(t, endpoint)) + wantFrame(t, frames, record(t, "<158>", "gitea", "first")) + wantFrame(t, frames, record(t, "<158>", "gitea", "second")) + }) +} + +func TestStalledEndpointHoldsUpNoWriteAndOldestRecordsAreDropped(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + certificate, roots := testCertificate(t) + endpoint := listen(t) + endpointParams := params(remotelog.SchemeTLS, endpoint.Addr()) + endpointParams.RootCAs = roots + endpointParams.Buffer = 3 + sender, _, _ := run(t, endpointParams) + + // The sender connects, and its TLS handshake waits for an answer + // the endpoint does not give yet. + conn := accept(t, endpoint) + + var stdout bytes.Buffer + + out := io.MultiWriter(&stdout, sender) + for i := range 5 { + _, _ = fmt.Fprintf(out, "line %d\n", i+1) + } + + if stdout.String() != "line 1\nline 2\nline 3\nline 4\nline 5\n" { + t.Errorf("stdout has %q", stdout.String()) + } + + wantCounts(t, sender, 0, 2, 3) + + // Once the endpoint answers, the three newest records are sent. + server := tls.Server(conn, &tls.Config{ + Certificates: []tls.Certificate{certificate}, + MinVersion: tls.VersionTLS12, + }) + + err := server.HandshakeContext(t.Context()) + if err != nil { + t.Fatalf("handshake: %v", err) + } + + synctest.Wait() + wantCounts(t, sender, 3, 2, 0) + + frames := bufio.NewReader(server) + for _, line := range []string{"line 3", "line 4", "line 5"} { + wantFrame(t, frames, record(t, local0Info, appName, line)) + } + }) +} + +func TestReconnectsWithBackoffAfterTheEndpointGoesAway(t *testing.T) { + t.Parallel() + + certificate, roots := testCertificate(t) + endpoint := startTLSEndpoint(t, certificate) + + synctest.Test(t, func(t *testing.T) { + endpointParams := params(remotelog.SchemeTLS, endpoint.addr) + endpointParams.RootCAs = roots + sender, logged, _ := run(t, endpointParams) + + _, _ = sender.Write([]byte("one\n")) + + synctest.Wait() + wantCounts(t, sender, 1, 0, 0) + + conn := endpoint.next(t) + wantFrame(t, bufio.NewReader(conn), record(t, local0Info, appName, "one")) + + // The endpoint goes away: it closes the connection, and refuses the + // next ones. The sender notices when a record fails, and tries to + // connect again a second later, then two seconds after that. + endpoint.refusing.Store(true) + + _ = conn.Close() + + writeUntilDropped(t, sender, 1) + sent := sender.Sent() + + _, _ = sender.Write([]byte("two\n")) + + time.Sleep(time.Second) + synctest.Wait() + + endpoint.refusing.Store(false) + + time.Sleep(2*time.Second - time.Nanosecond) + synctest.Wait() + wantCounts(t, sender, sent, 1, 1) + + // The endpoint is back, and the record waiting is sent. + time.Sleep(time.Nanosecond) + synctest.Wait() + wantCounts(t, sender, sent+1, 1, 0) + + conn = endpoint.next(t) + wantFrame(t, bufio.NewReader(conn), record(t, local0Info, appName, "two")) + wantRetries(t, logged, "1s", "2s") + }) +} + +func TestAConnectionClosedAtOnceIsMadeAgainAfterAGrowingDelay(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + endpoint := listen(t) + sender, logged, _ := run(t, params(remotelog.SchemeTCP, endpoint.Addr())) + + // The endpoint closes each connection as soon as it takes it. The + // sender notices when a record fails, and connects again a second + // later, then two seconds after that, then four. + delays := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} + for i, delay := range delays { + _ = accept(t, endpoint).Close() + + writeUntilDropped(t, sender, int64(i+1)) + wantConnectedAgainAfter(t, sender, delay) + } + + wantRetries(t, logged, "1s", "2s", "4s") + }) +} + +func TestTheDelayStartsAgainAfterAConnectionThatStayedUpAMinute(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + endpoint := listen(t) + sender, logged, _ := run(t, params(remotelog.SchemeTCP, endpoint.Addr())) + + _ = accept(t, endpoint).Close() + + writeUntilDropped(t, sender, 1) + wantConnectedAgainAfter(t, sender, time.Second) + + // A connection that fails just short of a minute after it was made + // leaves the delay growing. + conn := accept(t, endpoint) + + time.Sleep(time.Minute - time.Nanosecond) + + _ = conn.Close() + + writeUntilDropped(t, sender, 2) + wantConnectedAgainAfter(t, sender, 2*time.Second) + + // One that fails a minute after it was made starts it again from a + // second. + conn = accept(t, endpoint) + + time.Sleep(time.Minute) + + _ = conn.Close() + + writeUntilDropped(t, sender, 3) + wantConnectedAgainAfter(t, sender, time.Second) + wantRetries(t, logged, "1s", "2s", "1s") + }) +} + +func TestALineTooLongForADatagramIsDroppedAlone(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + endpoint, err := (&net.ListenConfig{}).ListenPacket(t.Context(), "udp", loopback) + if err != nil { + t.Fatalf("listen: %v", err) + } + + t.Cleanup(func() { _ = endpoint.Close() }) + + sender, logged, _ := run(t, params(remotelog.SchemeUDP, endpoint.LocalAddr())) + + // With its header, the first line's record is longer than the 65507 + // bytes a UDP datagram over IPv4 holds. It is dropped, nothing is + // logged, and the next line is sent at once. + _, _ = sender.Write([]byte(strings.Repeat("x", 65507) + "\nnext\n")) + + synctest.Wait() + wantCounts(t, sender, 1, 1, 0) + wantRetries(t, logged) + + datagram := make([]byte, 1024) + + n, _, err := endpoint.ReadFrom(datagram) + if err != nil { + t.Fatalf("read: %v", err) + } + + want := record(t, local0Info, appName, "next") + if string(datagram[:n]) != want { + t.Errorf("datagram %q, want %q", datagram[:n], want) + } + }) +} + +func TestRecordsWaitingAtTheStopAreSent(t *testing.T) { + t.Parallel() + + certificate, roots := testCertificate(t) + endpoint := startTLSEndpoint(t, certificate) + + synctest.Test(t, func(t *testing.T) { + // The endpoint refuses the sender's first connection: it fails to + // connect, and waits a second to try again. + endpoint.refusing.Store(true) + + endpointParams := params(remotelog.SchemeTLS, endpoint.addr) + endpointParams.RootCAs = roots + sender, logged, stop := run(t, endpointParams) + + synctest.Wait() + wantRetries(t, logged, "1s") + + _, _ = sender.Write([]byte("one\ntwo\n")) + + endpoint.refusing.Store(false) + + // Stopped before that second is over, it connects to send them. + stop() + wantCounts(t, sender, 2, 0, 0) + + frames := bufio.NewReader(endpoint.next(t)) + wantFrame(t, frames, record(t, local0Info, appName, "one")) + wantFrame(t, frames, record(t, local0Info, appName, "two")) + }) +} + +// output collects what the sender logs. +type output struct { + mu sync.Mutex + buf bytes.Buffer +} + +// Write adds lines the sender logs. +func (o *output) Write(p []byte) (int, error) { + o.mu.Lock() + defer o.mu.Unlock() + + return o.buf.Write(p) +} + +// text returns everything logged so far. +func (o *output) text() string { + o.mu.Lock() + defer o.mu.Unlock() + + return o.buf.String() +} + +// params returns the settings of a Sender for the endpoint at addr, in +// the form scheme names: room for ten lines, the default facility, and +// appName. +func params(scheme string, addr net.Addr) remotelog.Params { + return remotelog.Params{ + URL: &url.URL{Scheme: scheme, Host: addr.String()}, + Buffer: 10, + Facility: local0, + AppName: appName, + } +} + +// run runs a Sender with settings until the test ends or the function +// it returns is called, which waits for Run to return. It returns the +// Sender, and what it logs. +func run(t *testing.T, settings remotelog.Params) (*remotelog.Sender, *output, func()) { + t.Helper() + + sender := remotelog.New(settings) + logged := &output{} + ctx, cancel := context.WithCancel(t.Context()) + ran := make(chan struct{}) + + go func() { + sender.Run(ctx, slog.New(slog.NewJSONHandler(logged, nil))) + close(ran) + }() + + stop := func() { + cancel() + <-ran + } + t.Cleanup(stop) + + return sender, logged, stop +} + +// listen returns a TCP listener on the loopback address, closed when the +// test ends. +func listen(t *testing.T) net.Listener { + t.Helper() + + listener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", loopback) + if err != nil { + t.Fatalf("listen: %v", err) + } + + t.Cleanup(func() { _ = listener.Close() }) + + return listener +} + +// accept returns the next connection to listener, closed when the test +// ends. +func accept(t *testing.T, listener net.Listener) net.Conn { + t.Helper() + + conn, err := listener.Accept() + if err != nil { + t.Fatalf("accept: %v", err) + } + + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// tlsEndpoint is a syslog+tls endpoint on the loopback address, which a +// test starts outside its bubble. It keeps its listener until the test +// ends, and either takes each connection or refuses it. +type tlsEndpoint struct { + addr net.Addr + // refusing is set while the endpoint closes each connection before the + // TLS handshake, which fails the sender's attempt to connect. + refusing atomic.Bool + // conns are the connections it has taken, after the handshake. + conns chan net.Conn +} + +// startTLSEndpoint starts a tlsEndpoint with certificate, which takes +// connections until it is told to refuse them. +func startTLSEndpoint(t *testing.T, certificate tls.Certificate) *tlsEndpoint { + t.Helper() + + listener := listen(t) + endpoint := &tlsEndpoint{addr: listener.Addr(), conns: make(chan net.Conn, 10)} + config := &tls.Config{ + Certificates: []tls.Certificate{certificate}, + MinVersion: tls.VersionTLS12, + } + + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + + server := tls.Server(conn, config) + if endpoint.refusing.Load() || server.HandshakeContext(t.Context()) != nil { + _ = conn.Close() + + continue + } + + endpoint.conns <- server + } + }() + + return endpoint +} + +// next returns the next connection the endpoint has taken, closed when +// the test ends. +func (e *tlsEndpoint) next(t *testing.T) net.Conn { + t.Helper() + + conn := <-e.conns + + t.Cleanup(func() { _ = conn.Close() }) + + return conn +} + +// record returns the record of line made as the test started, with the +// priority and the app name given. +func record(t *testing.T, priority, app, line string) string { + t.Helper() + + hostname, err := os.Hostname() + if err != nil || hostname == "" { + hostname = "-" + } + + return priority + "1 " + started + " " + hostname + " " + app + " - - - " + line +} + +// wantFrame reads the next octet-counted frame from frames, and checks +// that it holds want. +func wantFrame(t *testing.T, frames *bufio.Reader, want string) { + t.Helper() + + count, err := frames.ReadString(' ') + if err != nil { + t.Fatalf("read a frame's length: %v", err) + } + + length, err := strconv.Atoi(strings.TrimSuffix(count, " ")) + if err != nil { + t.Fatalf("frame starts %q, not with its length", count) + } + + got := make([]byte, length) + + _, err = io.ReadFull(frames, got) + if err != nil { + t.Fatalf("read a frame: %v", err) + } + + if string(got) != want { + t.Errorf("frame %q, want %q", got, want) + } +} + +// wantCounts checks the records sender has sent, dropped and holds in +// its buffer. +func wantCounts( + t *testing.T, sender *remotelog.Sender, sent, dropped int64, depth int, +) { + t.Helper() + + if sender.Sent() != sent || sender.Dropped() != dropped || sender.Depth() != depth { + t.Fatalf("sent %d, dropped %d, %d in the buffer; want %d, %d and %d", + sender.Sent(), sender.Dropped(), sender.Depth(), sent, dropped, depth) + } +} + +// writeUntilDropped writes a line at a time until the count of records +// sender has dropped reaches dropped. The records it sends on a +// connection the endpoint has closed are lost before one fails; how many +// depends on when the endpoint's host answers that the connection is +// gone. +func writeUntilDropped(t *testing.T, sender *remotelog.Sender, dropped int64) { + t.Helper() + + for sender.Dropped() < dropped { + _, _ = sender.Write([]byte("lost\n")) + + synctest.Wait() + } +} + +// wantConnectedAgainAfter writes a line while the sender waits to connect +// again, and checks that it connects, and takes the line from the buffer, +// only once delay is over. +func wantConnectedAgainAfter( + t *testing.T, sender *remotelog.Sender, delay time.Duration, +) { + t.Helper() + + _, _ = sender.Write([]byte("waiting\n")) + + time.Sleep(delay - time.Nanosecond) + synctest.Wait() + + if sender.Depth() != 1 { + t.Fatalf("connected again before %v", delay) + } + + time.Sleep(time.Nanosecond) + synctest.Wait() + + if sender.Depth() != 0 { + t.Fatalf("not connected again after %v", delay) + } +} + +// wantRetries checks that the sender logged a failure, of an attempt to +// connect or of a connection, for each of delays, the time until the next +// attempt, in order, and logged nothing else. +func wantRetries(t *testing.T, logged *output, delays ...string) { + t.Helper() + + var got []string + + for line := range strings.Lines(logged.text()) { + var fields map[string]any + + err := json.Unmarshal([]byte(line), &fields) + if err != nil || fields["msg"] != "sending to SWWAF_LOG_REMOTE_URL failed" { + t.Fatalf("logged %q", line) + } + + delay, _ := fields["connecting_again_in"].(string) + got = append(got, delay) + } + + if !slices.Equal(got, delays) { + t.Errorf("logged failures to connect again in %v, want %v", got, delays) + } +} + +// testCertificate returns a certificate for 127.0.0.1 that is its own +// CA, and a pool that holds it. It is valid on the bubble's clock, which +// starts at 2000-01-01T00:00:00Z. +func testCertificate(t *testing.T) (tls.Certificate, *x509.CertPool) { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("generate a key: %v", err) + } + + template := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "smallwebwaf test CA"}, + NotBefore: time.Date(1999, 12, 31, 0, 0, 0, 0, time.UTC), + NotAfter: time.Date(2000, 1, 2, 0, 0, 0, 0, time.UTC), + IsCA: true, + BasicConstraintsValid: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1)}, + } + + der, err := x509.CreateCertificate(rand.Reader, template, template, + &key.PublicKey, key) + if err != nil { + t.Fatalf("create a certificate: %v", err) + } + + certificate, err := x509.ParseCertificate(der) + if err != nil { + t.Fatalf("parse the certificate: %v", err) + } + + roots := x509.NewCertPool() + roots.AddCert(certificate) + + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key}, roots +} diff --git a/internal/smallwebwaf/smallwebwaf.go b/internal/smallwebwaf/smallwebwaf.go index 32e56ee..d39cf59 100644 --- a/internal/smallwebwaf/smallwebwaf.go +++ b/internal/smallwebwaf/smallwebwaf.go @@ -18,6 +18,7 @@ import ( "sneak.berlin/go/smallwebwaf/internal/config" "sneak.berlin/go/smallwebwaf/internal/lookup" "sneak.berlin/go/smallwebwaf/internal/proxy" + "sneak.berlin/go/smallwebwaf/internal/remotelog" "sneak.berlin/go/smallwebwaf/internal/requestlog" "sneak.berlin/go/smallwebwaf/internal/rules" "sneak.berlin/go/smallwebwaf/internal/state" @@ -28,6 +29,11 @@ import ( // runit and docker wait a little longer before they kill the process. const shutdownTimeout = 5 * time.Second +// remoteLogStopTimeout is how long, as smallwebwaf stops, the log lines +// still waiting are sent to SWWAF_LOG_REMOTE_URL before they are given +// up. stdout has carried them. +const remoteLogStopTimeout = 2 * time.Second + // Params are what Run needs from the process. type Params struct { // Version is the version of the binary, set when it is built. @@ -70,6 +76,21 @@ func Run(ctx context.Context, params Params) int { return 1 } + // While SWWAF_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) + + stopSending := startSending(ctx, remote, processLog) + defer stopSending() + } + ruleFiles, err := rules.Load(rules.Params{ Dir: cfg.RulesDir, Enabled: cfg.RulesEnabled, @@ -86,12 +107,15 @@ func Run(ctx context.Context, params Params) int { server := proxy.New(proxy.Params{ Config: cfg, - RequestLog: params.Stdout, + RequestLog: stdout, ProcessLog: processLog, GeoJSURL: lookup.URL, Now: now, Rules: ruleFiles, }) + if remote != nil { + server.Metrics.AddRemoteLog(remote) + } files, err := state.Load(state.Params{ Dir: cfg.StateDir, @@ -126,6 +150,43 @@ func Run(ctx context.Context, params Params) int { return serve(ctx, server.Server, listener, files, ruleFiles, processLog) } +// newRemoteLogSender returns a sender of the log lines to +// SWWAF_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 smallwebwaf 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, and reads the rule files again // as they change, until ctx is done. Then it gives the requests in diff --git a/internal/smallwebwaf/smallwebwaf_internal_test.go b/internal/smallwebwaf/smallwebwaf_internal_test.go new file mode 100644 index 0000000..aea6bfe --- /dev/null +++ b/internal/smallwebwaf/smallwebwaf_internal_test.go @@ -0,0 +1,82 @@ +package smallwebwaf + +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 +} diff --git a/internal/smallwebwaf/smallwebwaf_test.go b/internal/smallwebwaf/smallwebwaf_test.go index fa5be79..fa669d0 100644 --- a/internal/smallwebwaf/smallwebwaf_test.go +++ b/internal/smallwebwaf/smallwebwaf_test.go @@ -10,6 +10,8 @@ import ( "net/http/httptest" "os" "path/filepath" + "slices" + "strconv" "strings" "sync" "testing" @@ -385,6 +387,112 @@ func TestRulesDirThatDoesNotExistStopsTheStart(t *testing.T) { ": no such file or directory") } +func TestEveryLineIsAlsoSentToTheRemoteLogEndpoint(t *testing.T) { + t.Parallel() + + endpoint, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + defer func() { + _ = endpoint.Close() + }() + + env := map[string]string{ + listenAddr: localhost + ":0", + upstreamURL: startApp(t), + stateDir: t.TempDir(), + rulesDir: t.TempDir(), + "SWWAF_LOG_REMOTE_URL": "syslog+tcp://" + endpoint.Addr().String(), + } + + out := runUntilStopped(t, env, func(url string) { + wantGreeting(t, url) + }) + out.line(t, "type", "request") + + // smallwebwaf connected as it started, and closes the connection once + // it has sent the lines written as it stopped. + conn, err := endpoint.Accept() + if err != nil { + t.Fatalf("accept: %v", err) + } + + received, err := io.ReadAll(conn) + _ = conn.Close() + + if err != nil { + t.Fatalf("read: %v", err) + } + + // Lines written at once by several goroutines may reach stdout and + // the endpoint in different orders. + sent := messages(t, string(received)) + written := slices.Collect(strings.Lines(out.text())) + + slices.Sort(sent) + slices.Sort(written) + + if !slices.Equal(sent, written) { + t.Errorf("sent\n%v\nwrote\n%v", sent, written) + } +} + +func TestStalledRemoteLogEndpointHoldsUpNoRequest(t *testing.T) { + t.Parallel() + + const token = "0123456789abcdef0123456789abcdef" + + // The endpoint takes connections and never answers, so the TLS + // handshake of each waits on it, and no line is ever sent. + endpoint, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", localhost+":0") + if err != nil { + t.Fatalf("listen: %v", err) + } + + defer func() { + _ = endpoint.Close() + }() + + env := map[string]string{ + listenAddr: localhost + ":0", + upstreamURL: startApp(t), + stateDir: t.TempDir(), + rulesDir: t.TempDir(), + "SWWAF_LOG_REMOTE_URL": "syslog+tls://" + endpoint.Addr().String(), + "SWWAF_LOG_REMOTE_BUFFER": "1", + "SWWAF_METRICS_TOKEN": token, + } + + out := runUntilStopped(t, env, func(url string) { + wantGreeting(t, url) + + // More than one line has been written, and the buffer holds the + // last. + metrics := metricsText(t, url+"_smallwebwaf/metrics", token) + for _, series := range []string{ + "smallwebwaf_remote_log_lines_sent_total 0", + "smallwebwaf_remote_log_buffer_depth 1", + } { + if !strings.Contains(metrics, "\n"+series+"\n") { + t.Errorf("no %q in the metrics:\n%s", series, metrics) + } + } + + if strings.Contains(metrics, "\nsmallwebwaf_remote_log_lines_dropped_total 0\n") || + !strings.Contains(metrics, "\nsmallwebwaf_remote_log_lines_dropped_total ") { + t.Errorf("no line dropped in the metrics:\n%s", metrics) + } + + // Closed, the endpoint refuses the connection made to send the + // lines still waiting at the stop, which then does not wait. + _ = endpoint.Close() + }) + + out.line(t, "type", "request") +} + func TestStateFileThatDoesNotParseStopsTheStart(t *testing.T) { t.Parallel() @@ -581,6 +689,69 @@ func wantGreeting(t *testing.T, url string) { } } +// messages returns the message of each record in received, octet-counted +// frames of RFC 5424 records with the default facility and app name, each +// with the newline that ends a line on stdout. +func messages(t *testing.T, received string) []string { + t.Helper() + + hostname, _ := os.Hostname() + header := " " + hostname + " " + hostname + " - - - " + + var found []string + + for received != "" { + count, rest, _ := strings.Cut(received, " ") + + length, err := strconv.Atoi(count) + if err != nil || length > len(rest) { + t.Fatalf("no frame at %q", received) + } + + record := rest[:length] + received = rest[length:] + + _, message, ok := strings.Cut(record, header) + if !ok || !strings.HasPrefix(record, "<134>1 ") { + t.Fatalf("record %q, want priority <134> and header %q", record, header) + } + + found = append(found, message+"\n") + } + + return found +} + +// metricsText asks for the metrics at url with token, and returns them. +func metricsText(t *testing.T, url, token string) string { + t.Helper() + + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, url, + http.NoBody) + if err != nil { + t.Fatalf("new request: %v", err) + } + + req.Header.Set("Authorization", "Bearer "+token) + + transport := &http.Transport{} + defer transport.CloseIdleConnections() + + res, err := (&http.Client{Transport: transport}).Do(req) + if err != nil { + t.Fatalf("request: %v", err) + } + + body, err := io.ReadAll(res.Body) + _ = res.Body.Close() + + if err != nil || res.StatusCode != http.StatusOK { + t.Fatalf("metrics answered %d (%v)", res.StatusCode, err) + } + + return string(body) +} + // wantRefused checks that a request to url is refused with 403, the // default SWWAF_BAN_RESPONSE. func wantRefused(t *testing.T, url string) {