smallwebwaf/internal/remotelog/remotelog_test.go

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/blog"
	// 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 = "blog"
		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>", "blog", "first"))
		wantFrame(t, frames, record(t, "<158>", "blog", "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
}