v0.20.0 mtls hardening, fail-closed config, startup warnings, sink fixes

internal/sink/console/console.go

@@ -65,7 +65,7 @@ func NewConsoleSinkPlugin(
 	opts := &config.ConsoleSinkOptions{}
 
 	// Scan config map into struct
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 
@@ -206,4 +206,4 @@ func (cs *ConsoleSink) processLoop(ctx context.Context) {
 			return
 		}
 	}
-}
+}

internal/sink/file/file.go

@@ -68,7 +68,7 @@ func NewFileSinkPlugin(
 	opts := &config.FileSinkOptions{}
 
 	// Scan config map into struct
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 
@@ -259,4 +259,4 @@ func (fs *FileSink) processLoop(ctx context.Context) {
 			return
 		}
 	}
-}
+}

internal/sink/http/http.go

@@ -107,7 +107,7 @@ func NewHTTPSinkPlugin(
 		Host:           DefaultHTTPHost,
 		WriteTimeoutMS: 0, // SSE indefinite streaming
 	}
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
@@ -167,12 +167,8 @@ func NewHTTPSinkPlugin(
 		"tls", tlsCfg != nil,
 		"mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert,
 		"auth", authPolicy.Describe())
-	if authPolicy.Unrestricted() {
-		logger.Warn("msg", "Auth policy admits any identity the configured CA vouches for",
-			"component", "http_sink",
-			"instance_id", id,
-			"hint", "set auth.allow or auth.allow_patterns to authorize named clients")
-	}
+	tlsx.LogWarnings(logger, "http_sink", id, opts.TLS, true)
+	authPolicy.LogStartup(logger, "http_sink", id, false)
 	return h, nil
 }
 
@@ -407,7 +403,10 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) {
 
 	w.Header().Set("Content-Type", "text/event-stream")
 	w.Header().Set("Cache-Control", "no-cache")
-	w.Header().Set("Access-Control-Allow-Origin", "*")
+	if !h.auth.Enabled() {
+		// An authenticated stream is not offered to every web origin
+		w.Header().Set("Access-Control-Allow-Origin", "*")
+	}
 	w.Header().Set("X-Accel-Buffering", "no")
 	w.WriteHeader(http.StatusOK)
 

internal/sink/http/http_test.go

@@ -2,13 +2,18 @@ package http
 
 import (
 	"bufio"
+	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"os"
+	"strings"
 	"testing"
 	"time"
 
 	"logwisp/internal/sink"
+	"logwisp/internal/testutil"
 )
 
 func TestStatusReportsQueueAndConnectionBounds(t *testing.T) {
@@ -119,3 +124,48 @@ func TestHeadOnStreamPathIsRefused(t *testing.T) {
 		t.Errorf("sessions after HEAD = %d, want 0", n)
 	}
 }
+
+// A stream gated by an auth policy is not offered to every web origin; an
+// open stream keeps the wildcard so browser dashboards still work.
+func TestWildcardCORSOnlyWithoutAuth(t *testing.T) {
+	pki := testutil.NewPKI(t, "viewer-01")
+	caPEM, err := os.ReadFile(pki.CA)
+	if err != nil {
+		t.Fatal(err)
+	}
+	roots := x509.NewCertPool()
+	roots.AppendCertsFromPEM(caPEM)
+	clientCert, err := tls.LoadX509KeyPair(pki.ClientCert, pki.ClientKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	open, _ := newTestHTTPSink(t, nil)
+	gated, _ := newTestHTTPSink(t, map[string]any{
+		"tls": map[string]any{"enabled": true, "cert_file": pki.ServerCert, "key_file": pki.ServerKey,
+			"client_auth": true, "client_ca_file": pki.CA},
+		"auth": map[string]any{"type": "mtls", "allow": []any{"viewer-01"}},
+	})
+	for _, tc := range []struct {
+		name string
+		sink *HTTPSink
+		want string
+	}{{"open", open, "*"}, {"gated", gated, ""}} {
+		client, baseURL := serveTestHTTPSink(t, tc.sink)
+		if tc.sink.tlsConfig != nil {
+			client.Transport = &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots, Certificates: []tls.Certificate{clientCert}}}
+			baseURL = "https" + strings.TrimPrefix(baseURL, "http")
+		}
+		resp, err := client.Get(baseURL + "/stream")
+		if err != nil {
+			t.Fatalf("%s: %v", tc.name, err)
+		}
+		resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			t.Fatalf("%s: GET /stream = %d", tc.name, resp.StatusCode)
+		}
+		if got := resp.Header.Get("Access-Control-Allow-Origin"); got != tc.want {
+			t.Errorf("%s: Access-Control-Allow-Origin = %q, want %q", tc.name, got, tc.want)
+		}
+	}
+}

internal/sink/httpchain/httpchain.go

@@ -44,6 +44,7 @@ const (
 	DefaultHTTPChainSinkRequestTimeoutMS = 10000
 	DefaultHTTPChainSinkBackoffMinMS     = 500
 	DefaultHTTPChainSinkBackoffMaxMS     = 30000
+	maxResponseDrain                     = 64 * 1024
 )
 
 // HTTPChainSink batches structured entries and posts NDJSON to a downstream
@@ -92,7 +93,7 @@ func NewHTTPChainSinkPlugin(
 	proxy *session.Proxy,
 ) (sink.Sink, error) {
 	opts := &config.HTTPChainSinkOptions{}
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.NonEmpty(opts.Host); err != nil {
@@ -182,7 +183,11 @@ func NewHTTPChainSinkPlugin(
 		mtls:       tlsCfg != nil && len(tlsCfg.Certificates) > 0,
 		auth:       authPolicy,
 		url:        scheme + "://" + addr + opts.IngestPath,
-		client:     &http.Client{Transport: transport},
+		// A redirect would resend the batch, even from https to plaintext http
+		client: &http.Client{
+			Transport:     transport,
+			CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+		},
 		input:      make(chan core.TransportEvent, opts.BufferSize),
 		done:       make(chan struct{}),
 		logger:     logger,
@@ -208,6 +213,7 @@ func NewHTTPChainSinkPlugin(
 		"tls", t.tlsEnabled,
 		"mtls", t.mtls,
 		"auth", authPolicy.Describe())
+	tlsx.LogWarnings(logger, "http_chain_sink", id, opts.TLS, false)
 	return t, nil
 }
 
@@ -436,8 +442,8 @@ func (t *HTTPChainSink) post(ctx context.Context, body []byte) (transient bool,
 		return true, err
 	}
 	defer resp.Body.Close()
-	// Drain for connection reuse
-	io.Copy(io.Discard, resp.Body)
+	// Drain for connection reuse, bounded so a hostile peer cannot stream forever
+	io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponseDrain))
 
 	switch {
 	case resp.StatusCode >= 200 && resp.StatusCode < 300:

internal/sink/httpchain/httpchain_test.go

@@ -0,0 +1,47 @@
+package httpchain
+
+import (
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"logwisp/internal/session"
+
+	"github.com/lixenwraith/log"
+)
+
+// A redirect is never followed: it would resend the batch wherever the
+// response points, plaintext http included.
+func TestRedirectIsNotFollowed(t *testing.T) {
+	var leaked atomic.Bool
+	target := httptest.NewServer(http.HandlerFunc(func(http.ResponseWriter, *http.Request) { leaked.Store(true) }))
+	t.Cleanup(target.Close)
+	redirector := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Redirect(w, r, target.URL+r.URL.Path, http.StatusPermanentRedirect)
+	}))
+	t.Cleanup(redirector.Close)
+
+	host, port, err := net.SplitHostPort(redirector.Listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	p, _ := strconv.ParseInt(port, 10, 64)
+	manager := session.NewManager(time.Hour)
+	t.Cleanup(manager.Stop)
+	created, err := NewHTTPChainSinkPlugin("fwd", map[string]any{"host": host, "port": p},
+		log.NewLogger(), session.NewProxy(manager, "fwd"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	transient, err := created.(*HTTPChainSink).post(t.Context(), []byte("{}\n"))
+	if err == nil || transient {
+		t.Fatalf("post = (transient %v, %v), want a permanent error", transient, err)
+	}
+	if leaked.Load() {
+		t.Fatal("the batch followed the redirect")
+	}
+}

internal/sink/null/null.go

@@ -6,6 +6,7 @@ import (
 	"sync/atomic"
 	"time"
 
+	"logwisp/internal/config"
 	"logwisp/internal/core"
 	"logwisp/internal/plugin"
 	"logwisp/internal/session"
@@ -49,6 +50,9 @@ func NewNullSinkPlugin(
 	logger *log.Logger,
 	proxy *session.Proxy,
 ) (sink.Sink, error) {
+	if err := config.Scan(configMap, &config.NullSinkOptions{}); err != nil {
+		return nil, fmt.Errorf("failed to parse config: %w", err)
+	}
 	ns := &NullSink{
 		id:     id,
 		proxy:  proxy,
@@ -143,4 +147,4 @@ func (ns *NullSink) processLoop(ctx context.Context) {
 			return
 		}
 	}
-}
+}

internal/sink/tcp/tcp.go

@@ -58,8 +58,10 @@ type TCPSink struct {
 	input  chan core.TransportEvent
 	logger *log.Logger
 
-	// Client registry
+	// Client registry. conns holds every accepted connection, so shutdown also
+	// reaches peers still in the handshake, before they become clients.
 	clients      map[uint64]*tcpClient
+	conns        map[net.Conn]struct{}
 	clientsMu    sync.Mutex
 	nextClientID atomic.Uint64
 	writeTimeout time.Duration
@@ -86,11 +88,10 @@ type TCPSink struct {
 	lastProcessed  atomic.Value // time.Time
 }
 
-// tcpClient pairs a connection with its bounded send queue.
+// tcpClient is a registered connection's bounded send queue.
 // send is written by the broadcast loop (non-blocking) and drained by the
 // writer goroutine; closed signals reader-detected disconnect.
 type tcpClient struct {
-	conn      net.Conn
 	send      chan []byte
 	sessionID string
 	closed    chan struct{}
@@ -107,7 +108,7 @@ func NewTCPSinkPlugin(
 		Host:      DefaultTCPHost,
 		KeepAlive: true,
 	}
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.Port(opts.Port); err != nil {
@@ -143,6 +144,7 @@ func NewTCPSinkPlugin(
 		done:         make(chan struct{}),
 		logger:       logger,
 		clients:      make(map[uint64]*tcpClient),
+		conns:        make(map[net.Conn]struct{}),
 		writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond,
 		tlsConfig:    tlsCfg,
 		auth:         authPolicy,
@@ -157,12 +159,8 @@ func NewTCPSinkPlugin(
 		"tls", tlsCfg != nil,
 		"mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert,
 		"auth", authPolicy.Describe())
-	if authPolicy.Unrestricted() {
-		logger.Warn("msg", "Auth policy admits any identity the configured CA vouches for",
-			"component", "tcp_sink",
-			"instance_id", id,
-			"hint", "set auth.allow or auth.allow_patterns to authorize named clients")
-	}
+	tlsx.LogWarnings(logger, "tcp_sink", id, opts.TLS, true)
+	authPolicy.LogStartup(logger, "tcp_sink", id, false)
 	return t, nil
 }
 
@@ -258,8 +256,8 @@ func (t *TCPSink) shutdown() {
 			t.listener.Close() // unblocks acceptLoop
 		}
 		t.clientsMu.Lock()
-		for _, c := range t.clients {
-			c.conn.Close() // unblocks per-connection readers
+		for conn := range t.conns {
+			conn.Close() // unblocks handshakes and per-connection readers
 		}
 		t.clientsMu.Unlock()
 	})
@@ -287,9 +285,17 @@ func (t *TCPSink) acceptLoop() {
 			continue
 		}
 
-		// Certificate authorization runs in handleConn post-handshake,
-		// pre-registration. Password-auth extension point: preamble
-		// verification belongs at the same place.
+		t.clientsMu.Lock()
+		select {
+		case <-t.done:
+			// shutdown already swept conns; this one would outlive it
+			t.clientsMu.Unlock()
+			conn.Close()
+			return
+		default:
+		}
+		t.conns[conn] = struct{}{}
+		t.clientsMu.Unlock()
 
 		t.wg.Add(1)
 		go t.handleConn(conn)
@@ -305,6 +311,10 @@ func (t *TCPSink) handleConn(conn net.Conn) {
 	// Counted from accept: max_connections bounds concurrent handshakes too
 	count := t.activeConns.Add(1)
 	defer func() {
+		conn.Close()
+		t.clientsMu.Lock()
+		delete(t.conns, conn)
+		t.clientsMu.Unlock()
 		newCount := t.activeConns.Add(-1)
 		t.logger.Debug("msg", "TCP connection closed",
 			"component", "tcp_sink",
@@ -327,7 +337,6 @@ func (t *TCPSink) handleConn(conn net.Conn) {
 				"component", "tcp_sink",
 				"remote_addr", remote,
 				"error", err)
-			conn.Close()
 			return
 		}
 		cs := tc.ConnectionState()
@@ -348,14 +357,12 @@ func (t *TCPSink) handleConn(conn net.Conn) {
 			"instance_id", t.id,
 			"remote_addr", remote,
 			"error", err)
-		conn.Close()
 		return
 	}
 	ident.Apply(meta)
 
 	sess := t.proxy.CreateSession(remote, meta)
 	c := &tcpClient{
-		conn:      conn,
 		send:      make(chan []byte, t.config.ClientBufferSize),
 		sessionID: sess.ID,
 		closed:    make(chan struct{}),

internal/sink/tcp/tcp_test.go

@@ -0,0 +1,59 @@
+package tcp
+
+import (
+	"net"
+	"strconv"
+	"testing"
+	"time"
+
+	"logwisp/internal/session"
+	"logwisp/internal/testutil"
+
+	"github.com/lixenwraith/log"
+)
+
+// A peer that connects and never completes the TLS handshake does not hold
+// Stop for the handshake timeout: shutdown reaches connections before they
+// become clients.
+func TestStopDoesNotWaitForSilentHandshake(t *testing.T) {
+	pki := testutil.NewPKI(t, "viewer")
+	probe, err := net.Listen("tcp4", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	addr := probe.Addr().String()
+	probe.Close()
+	_, port, _ := net.SplitHostPort(addr)
+	p, _ := strconv.ParseInt(port, 10, 64)
+
+	manager := session.NewManager(time.Hour)
+	t.Cleanup(manager.Stop)
+	created, err := NewTCPSinkPlugin("out", map[string]any{
+		"host": "127.0.0.1",
+		"port": p,
+		"tls":  map[string]any{"enabled": true, "cert_file": pki.ServerCert, "key_file": pki.ServerKey},
+	}, log.NewLogger(), session.NewProxy(manager, "out"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	sink := created.(*TCPSink)
+	if err := sink.Start(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	conn, err := net.Dial("tcp4", addr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	for deadline := time.Now().Add(2 * time.Second); sink.activeConns.Load() == 0; time.Sleep(10 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("connection never reached the handshake")
+		}
+	}
+
+	start := time.Now()
+	sink.Stop()
+	if d := time.Since(start); d > 2*time.Second {
+		t.Fatalf("Stop took %v with a silent peer in the handshake", d)
+	}
+}

internal/sink/tcpchain/tcpchain.go

@@ -6,7 +6,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"maps"
-	"math/rand/v2"
 	"net"
 	"os"
 	"strconv"
@@ -88,7 +87,7 @@ func NewTCPChainSinkPlugin(
 	opts := &config.TCPChainSinkOptions{
 		KeepAlive: true,
 	}
-	if err := lconfig.ScanMap(configMap, opts); err != nil {
+	if err := config.Scan(configMap, opts); err != nil {
 		return nil, fmt.Errorf("failed to parse config: %w", err)
 	}
 	if err := lconfig.NonEmpty(opts.Host); err != nil {
@@ -179,6 +178,7 @@ func NewTCPChainSinkPlugin(
 		"tls", tlsCfg != nil,
 		"mtls", tlsCfg != nil && len(tlsCfg.Certificates) > 0,
 		"auth", authPolicy.Describe())
+	tlsx.LogWarnings(logger, "tcp_chain_sink", id, opts.TLS, false)
 	return t, nil
 }
 
@@ -299,24 +299,6 @@ func (t *TCPChainSink) runLoop(ctx context.Context) {
 	}
 }
 
-// toEntry extracts the structured entry, stamping node identity at first hop
-func (t *TCPChainSink) toEntry(event core.TransportEvent) core.LogEntry {
-	entry := event.Entry
-	if entry.Time.IsZero() {
-		// Defensive: event without structured entry, wrap formatted payload
-		t.synthesized.Add(1)
-		entry = core.LogEntry{
-			Time:    event.Time,
-			Source:  t.id,
-			Message: string(event.Payload),
-		}
-	}
-	if entry.Node == "" {
-		entry.Node = t.node
-	}
-	return entry
-}
-
 // deliver writes one line, holding it across reconnects until sent or shutdown.
 // Backpressure during outage propagates to the pipeline dispatch drop counter.
 func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool {
@@ -427,17 +409,3 @@ func (t *TCPChainSink) waitBackoff(ctx context.Context, failures int) bool {
 		return false
 	}
 }
-
-// backoffDelay computes exponential backoff with ±20% jitter
-func (t *TCPChainSink) backoffDelay(failures int) time.Duration {
-	minD := time.Duration(t.config.BackoffMinMS) * time.Millisecond
-	maxD := time.Duration(t.config.BackoffMaxMS) * time.Millisecond
-
-	d := maxD
-	if failures < 63 {
-		if v := minD << uint(failures-1); v > 0 && v < maxD {
-			d = v
-		}
-	}
-	return d - d/5 + time.Duration(rand.Int64N(int64(2*d/5)+1))
-}