From 296b3518836f9fbcf661b2fac8954971bea70ccc9c079159d11aa84475ca3618 Mon Sep 17 00:00:00 2001 From: Lixen Wraith Date: Sat, 18 Jul 2026 04:10:22 -0400 Subject: [PATCH] v0.16.0 tls added to tcp/http sources and sinks --- internal/config/config.go | 149 +++++++++++++++---------- internal/sink/http/http.go | 57 +++++++--- internal/sink/httpchain/httpchain.go | 47 ++++++-- internal/sink/tcp/tcp.go | 102 ++++++++++++----- internal/sink/tcpchain/tcpchain.go | 46 ++++++-- internal/source/httpchain/httpchain.go | 52 +++++++-- internal/source/tcpchain/tcpchain.go | 95 ++++++++++------ internal/tlsx/tlsx.go | 143 ++++++++++++++++++++++++ 8 files changed, 526 insertions(+), 165 deletions(-) create mode 100644 internal/tlsx/tlsx.go diff --git a/internal/config/config.go b/internal/config/config.go index 617d442..e531f36 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -208,27 +208,29 @@ type ConsoleSourceOptions struct { // TCPChainSourceOptions defines settings for a stdlib TCP listener ingesting // NDJSON entries from upstream logwisp tcp_chain sinks type TCPChainSourceOptions struct { - Host string `toml:"host"` - Port int64 `toml:"port"` - BufferSize int64 `toml:"buffer_size"` - MaxConnections int64 `toml:"max_connections"` // 0 = unlimited - ReadTimeoutMS int64 `toml:"read_timeout_ms"` // per-connection idle deadline, 0 = none - HelloTimeoutMS int64 `toml:"hello_timeout_ms"` // preamble deadline - TrustNode bool `toml:"trust_node"` // false: force node label from remote address - // Future: TLS/auth options + TLS *TLSOptions `toml:"tls"` + Host string `toml:"host"` + Port int64 `toml:"port"` + BufferSize int64 `toml:"buffer_size"` + MaxConnections int64 `toml:"max_connections"` // 0 = unlimited + ReadTimeoutMS int64 `toml:"read_timeout_ms"` // per-connection idle deadline, 0 = none + HelloTimeoutMS int64 `toml:"hello_timeout_ms"` // preamble deadline + TrustNode bool `toml:"trust_node"` // false: force node label from remote address + // Future: password auth block } // HTTPChainSourceOptions defines settings for a stdlib HTTP listener ingesting // NDJSON batches from upstream logwisp http_chain sinks type HTTPChainSourceOptions struct { - Host string `toml:"host"` - Port int64 `toml:"port"` - IngestPath string `toml:"ingest_path"` - BufferSize int64 `toml:"buffer_size"` - MaxBodyBytes int64 `toml:"max_body_bytes"` // per-request cap - ReadTimeoutMS int64 `toml:"read_timeout_ms"` // full request read deadline - TrustNode bool `toml:"trust_node"` // false: force node label from remote address - // Future: TLS/auth options + TLS *TLSOptions `toml:"tls"` + Host string `toml:"host"` + Port int64 `toml:"port"` + IngestPath string `toml:"ingest_path"` + BufferSize int64 `toml:"buffer_size"` + MaxBodyBytes int64 `toml:"max_body_bytes"` // per-request cap + ReadTimeoutMS int64 `toml:"read_timeout_ms"` // full request read deadline + TrustNode bool `toml:"trust_node"` // false: force node label from remote address + // Future: password auth block } // --- Sink Options --- @@ -273,59 +275,92 @@ type FileSinkOptions struct { // TCPSinkOptions defines settings for a TCP server sink type TCPSinkOptions struct { - Host string `toml:"host"` - Port int64 `toml:"port"` - BufferSize int64 `toml:"buffer_size"` // sink input queue - ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue - WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-write deadline - KeepAlive bool `toml:"keep_alive"` - KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"` - MaxConnections int64 `toml:"max_connections"` // 0 = unlimited - // Future: TLS (cert_file/key_file/client_ca), auth (token/mTLS) blocks + TLS *TLSOptions `toml:"tls"` + Host string `toml:"host"` + Port int64 `toml:"port"` + BufferSize int64 `toml:"buffer_size"` // sink input queue + ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue + WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-write deadline + KeepAlive bool `toml:"keep_alive"` + KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"` + MaxConnections int64 `toml:"max_connections"` // 0 = unlimited + // Future: password auth block } // HTTPSinkOptions defines settings for an HTTP SSE server sink type HTTPSinkOptions struct { - Host string `toml:"host"` - Port int64 `toml:"port"` - StreamPath string `toml:"stream_path"` - StatusPath string `toml:"status_path"` - BufferSize int64 `toml:"buffer_size"` // sink input queue - ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue - WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-SSE-write deadline, 0 = none - MaxConnections int64 `toml:"max_connections"` // 0 = unlimited - // Future: TLS (server.TLSConfig), auth middleware options + TLS *TLSOptions `toml:"tls"` + Host string `toml:"host"` + Port int64 `toml:"port"` + StreamPath string `toml:"stream_path"` + StatusPath string `toml:"status_path"` + BufferSize int64 `toml:"buffer_size"` // sink input queue + ClientBufferSize int64 `toml:"client_buffer_size"` // per-client send queue + WriteTimeoutMS int64 `toml:"write_timeout_ms"` // per-SSE-write deadline, 0 = none + MaxConnections int64 `toml:"max_connections"` // 0 = unlimited + // Future: password auth block } // TCPChainSinkOptions defines settings for a stdlib TCP client forwarding // entries to a downstream logwisp tcp_chain source type TCPChainSinkOptions struct { - Node string `toml:"node"` // origin label, default: os.Hostname() - Host string `toml:"host"` - Port int64 `toml:"port"` - BufferSize int64 `toml:"buffer_size"` - DialTimeoutMS int64 `toml:"dial_timeout_ms"` - WriteTimeoutMS int64 `toml:"write_timeout_ms"` - BackoffMinMS int64 `toml:"backoff_min_ms"` - BackoffMaxMS int64 `toml:"backoff_max_ms"` - KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"` - KeepAlive bool `toml:"keep_alive"` - // Future: TLS/auth options + TLS *TLSOptions `toml:"tls"` + Node string `toml:"node"` // origin label, default: os.Hostname() + Host string `toml:"host"` + Port int64 `toml:"port"` + BufferSize int64 `toml:"buffer_size"` + DialTimeoutMS int64 `toml:"dial_timeout_ms"` + WriteTimeoutMS int64 `toml:"write_timeout_ms"` + BackoffMinMS int64 `toml:"backoff_min_ms"` + BackoffMaxMS int64 `toml:"backoff_max_ms"` + KeepAlivePeriodMS int64 `toml:"keep_alive_period_ms"` + KeepAlive bool `toml:"keep_alive"` + // Future: password auth block } // HTTPChainSinkOptions defines settings for a stdlib HTTP client posting // NDJSON batches to a downstream logwisp http_chain source type HTTPChainSinkOptions struct { - Node string `toml:"node"` // origin label, default: os.Hostname() - Host string `toml:"host"` - Port int64 `toml:"port"` - IngestPath string `toml:"ingest_path"` - BufferSize int64 `toml:"buffer_size"` - MaxBatchCount int64 `toml:"max_batch_count"` - MaxBatchBytes int64 `toml:"max_batch_bytes"` - FlushIntervalMS int64 `toml:"flush_interval_ms"` - RequestTimeoutMS int64 `toml:"request_timeout_ms"` // covers dial + write + response - BackoffMinMS int64 `toml:"backoff_min_ms"` - BackoffMaxMS int64 `toml:"backoff_max_ms"` - // Future: TLS/auth options + TLS *TLSOptions `toml:"tls"` + Node string `toml:"node"` // origin label, default: os.Hostname() + Host string `toml:"host"` + Port int64 `toml:"port"` + IngestPath string `toml:"ingest_path"` + BufferSize int64 `toml:"buffer_size"` + MaxBatchCount int64 `toml:"max_batch_count"` + MaxBatchBytes int64 `toml:"max_batch_bytes"` + FlushIntervalMS int64 `toml:"flush_interval_ms"` + RequestTimeoutMS int64 `toml:"request_timeout_ms"` // covers dial + write + response + BackoffMinMS int64 `toml:"backoff_min_ms"` + BackoffMaxMS int64 `toml:"backoff_max_ms"` + // Future: password auth block +} + +// --- TLS Options --- + +// TLSOptions defines transport security for network sources and sinks. +// One shape serves both roles so the config block is uniform: +// - Listeners (tcp/http sinks, tcp_chain/http_chain sources) use +// cert_file/key_file as server identity; client_auth/client_ca_file +// require and verify peer certificates (mTLS). +// - Dialers (tcp_chain/http_chain sinks) use ca_file/server_name to verify +// the server; cert_file/key_file present a client identity (mTLS). +type TLSOptions struct { + Enabled bool `toml:"enabled"` + + // Local identity: required for listeners, optional for dialers (mTLS) + CertFile string `toml:"cert_file"` + KeyFile string `toml:"key_file"` + + // Listener-side peer verification (mTLS) + ClientAuth bool `toml:"client_auth"` + ClientCAFile string `toml:"client_ca_file"` + + // Dialer-side peer verification + CAFile string `toml:"ca_file"` // empty = system trust store + ServerName string `toml:"server_name"` // default: config host + InsecureSkipVerify bool `toml:"insecure_skip_verify"` + + // Minimum protocol version: "1.2" | "1.3" (default "1.3") + MinVersion string `toml:"min_version"` } diff --git a/internal/sink/http/http.go b/internal/sink/http/http.go index 6b5b0d3..d8274b1 100644 --- a/internal/sink/http/http.go +++ b/internal/sink/http/http.go @@ -2,6 +2,7 @@ package http import ( "context" + "crypto/tls" "encoding/json" "errors" "fmt" @@ -18,6 +19,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/sink" + "logwisp/internal/tlsx" "logwisp/internal/version" lconfig "github.com/lixenwraith/config" @@ -63,6 +65,10 @@ type HTTPSink struct { clients map[uint64]*sseClient clientsMu sync.Mutex nextClientID atomic.Uint64 + writeTimeout time.Duration + + // TLS + tlsConfig *tls.Config // Runtime done chan struct{} @@ -70,8 +76,6 @@ type HTTPSink struct { wg sync.WaitGroup startTime time.Time - writeTimeout time.Duration - // Statistics activeClients atomic.Int64 totalProcessed atomic.Uint64 @@ -122,6 +126,10 @@ func NewHTTPSinkPlugin( if opts.ClientBufferSize <= 0 { opts.ClientBufferSize = DefaultHTTPClientBufferSize } + tlsCfg, err := tlsx.Server(opts.TLS) + if err != nil { + return nil, err + } h := &HTTPSink{ id: id, @@ -133,6 +141,7 @@ func NewHTTPSinkPlugin( logger: logger, clients: make(map[uint64]*sseClient), writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, + tlsConfig: tlsCfg, } h.lastProcessed.Store(time.Time{}) @@ -142,17 +151,22 @@ func NewHTTPSinkPlugin( "host", opts.Host, "port", opts.Port, "stream_path", opts.StreamPath, - "status_path", opts.StatusPath) + "status_path", opts.StatusPath, + "tls", tlsCfg != nil, + "mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert) return h, nil } // Capabilities returns supported capabilities func (h *HTTPSink) Capabilities() []core.Capability { - // CapTLS/CapAuth appended when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, + caps := []core.Capability{core.CapSessionAware, core.CapMultiSession} + if h.tlsConfig != nil { + caps = append(caps, core.CapTLS) + if h.tlsConfig.ClientAuth == tls.RequireAndVerifyClientCert { + caps = append(caps, core.CapAuth) // mTLS is authentication + } } + return caps } // Input returns the channel for sending transport events @@ -163,8 +177,8 @@ func (h *HTTPSink) Input() chan<- core.TransportEvent { // Start binds the listener and serves stream/status endpoints func (h *HTTPSink) Start(ctx context.Context) error { // IPv4-only, parity with existing network sinks. - // TLS extension point: wrap ln with tls.NewListener (or set - // server.TLSConfig and use ServeTLS); single seam, handlers unchanged. + // TLS is applied via server.TLSConfig + ServeTLS below, not by wrapping + // ln; net/http then owns handshake, ALPN (h2), and per-conn errors. ln, err := net.Listen("tcp4", h.addr) if err != nil { return fmt.Errorf("http sink bind %s: %w", h.addr, err) @@ -183,15 +197,23 @@ func (h *HTTPSink) Start(ctx context.Context) error { Handler: handler, ReadHeaderTimeout: HTTPReadHeaderTimeout, // WriteTimeout unset by design: SSE responses are long-lived. - // Per-write deadlines via ResponseController in handleStream. + // net/http bounds the TLS handshake by min(ReadHeaderTimeout, + // ReadTimeout, WriteTimeout), so ReadHeaderTimeout covers it here. + ErrorLog: tlsx.HTTPErrorLog(h.logger, "http_sink"), } h.startTime = time.Now() h.wg.Add(1) go h.brokerLoop(ctx) + serve := h.server.Serve + if h.tlsConfig != nil { + h.server.TLSConfig = h.tlsConfig + serve = func(l net.Listener) error { return h.server.ServeTLS(l, "", "") } + } + go func() { - if err := h.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + if err := serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { h.logger.Error("msg", "HTTP server terminated", "component", "http_sink", "instance_id", h.id, @@ -312,9 +334,16 @@ func (h *HTTPSink) handleStream(w http.ResponseWriter, r *http.Request) { rc := http.NewResponseController(w) remote := r.RemoteAddr - sess := h.proxy.CreateSession(remote, map[string]any{ + meta := map[string]any{ "type": "http_client", - }) + } + if r.TLS != nil { + meta["tls"] = true + if cn := tlsx.PeerCN(*r.TLS); cn != "" { + meta["tls_peer_cn"] = cn + } + } + sess := h.proxy.CreateSession(remote, meta) c := &sseClient{ send: make(chan []byte, h.config.ClientBufferSize), @@ -402,6 +431,7 @@ func (h *HTTPSink) handleStatus(w http.ResponseWriter, r *http.Request) { "type": "http", "host": h.config.Host, "port": h.config.Port, + "tls": h.tlsConfig != nil, "active_clients": h.activeClients.Load(), "buffer_size": h.config.BufferSize, "uptime_seconds": int(time.Since(h.startTime).Seconds()), @@ -435,6 +465,7 @@ func (h *HTTPSink) GetStats() sink.SinkStats { "host": h.config.Host, "port": h.config.Port, "buffer_size": h.config.BufferSize, + "tls": h.tlsConfig != nil, "dropped_writes": h.droppedWrites.Load(), "rejected_clients": h.rejectedClients.Load(), "endpoints": map[string]string{ diff --git a/internal/sink/httpchain/httpchain.go b/internal/sink/httpchain/httpchain.go index 5dbbfaf..7c57b6b 100644 --- a/internal/sink/httpchain/httpchain.go +++ b/internal/sink/httpchain/httpchain.go @@ -21,6 +21,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/sink" + "logwisp/internal/tlsx" lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" @@ -54,6 +55,9 @@ type HTTPChainSink struct { node string url string + tlsEnabled bool + mtls bool + client *http.Client input chan core.TransportEvent logger *log.Logger @@ -128,6 +132,11 @@ func NewHTTPChainSinkPlugin( } } + tlsCfg, err := tlsx.Client(opts.TLS, opts.Host) + if err != nil { + return nil, err + } + addr := net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)) transport := &http.Transport{ @@ -139,16 +148,25 @@ func NewHTTPChainSinkPlugin( MaxIdleConnsPerHost: 2, IdleConnTimeout: 90 * time.Second, DisableCompression: true, - // Future: TLSClientConfig; HTTP/2 via ALPN once TLS lands + TLSClientConfig: tlsCfg, // nil = plaintext + TLSHandshakeTimeout: tlsx.HandshakeTimeout, + // h2 stays off: custom DialContext disables auto-ALPN and batched + // NDJSON POSTs gain nothing from it + } + + scheme := "http" + if tlsCfg != nil { + scheme = "https" } t := &HTTPChainSink{ - id: id, - proxy: proxy, - config: opts, - node: node, - // Future: "https" scheme with TLS - url: "http://" + addr + opts.IngestPath, + id: id, + proxy: proxy, + config: opts, + node: node, + tlsEnabled: tlsCfg != nil, + mtls: tlsCfg != nil && len(tlsCfg.Certificates) > 0, + url: scheme + "://" + addr + opts.IngestPath, client: &http.Client{Transport: transport}, input: make(chan core.TransportEvent, opts.BufferSize), done: make(chan struct{}), @@ -171,16 +189,22 @@ func NewHTTPChainSinkPlugin( "component", "http_chain_sink", "instance_id", id, "target", t.url, - "node", node) + "node", node, + "tls", t.tlsEnabled, + "mtls", t.mtls) return t, nil } // Capabilities returns supported capabilities func (t *HTTPChainSink) Capabilities() []core.Capability { - // CapTLS/CapAuth added when transport security lands - return []core.Capability{ - core.CapSessionAware, + caps := []core.Capability{core.CapSessionAware} + if t.tlsEnabled { + caps = append(caps, core.CapTLS) + if t.mtls { + caps = append(caps, core.CapAuth) // presents client identity (mTLS) + } } + return caps } // Input returns the channel for sending transport events @@ -234,6 +258,7 @@ func (t *HTTPChainSink) GetStats() sink.SinkStats { Details: map[string]any{ "target": t.url, "node": t.node, + "tls": t.tlsEnabled, "batches_sent": t.batchesSent.Load(), "request_errors": t.requestErrors.Load(), "dropped_batches": t.droppedBatches.Load(), diff --git a/internal/sink/tcp/tcp.go b/internal/sink/tcp/tcp.go index b20c10e..6b7db52 100644 --- a/internal/sink/tcp/tcp.go +++ b/internal/sink/tcp/tcp.go @@ -2,6 +2,7 @@ package tcp import ( "context" + "crypto/tls" "errors" "fmt" "net" @@ -15,6 +16,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/sink" + "logwisp/internal/tlsx" lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" @@ -58,6 +60,11 @@ type TCPSink struct { clients map[uint64]*tcpClient clientsMu sync.Mutex nextClientID atomic.Uint64 + writeTimeout time.Duration + + // TLS + tlsConfig *tls.Config + tlsHandshakeErrors atomic.Uint64 // Runtime done chan struct{} @@ -65,8 +72,6 @@ type TCPSink struct { wg sync.WaitGroup startTime time.Time - writeTimeout time.Duration - // Statistics activeConns atomic.Int64 totalProcessed atomic.Uint64 @@ -115,6 +120,10 @@ func NewTCPSinkPlugin( if opts.KeepAlivePeriodMS <= 0 { opts.KeepAlivePeriodMS = DefaultTCPKeepAlivePeriodMS } + tlsCfg, err := tlsx.Server(opts.TLS) + if err != nil { + return nil, err + } t := &TCPSink{ id: id, @@ -126,6 +135,7 @@ func NewTCPSinkPlugin( logger: logger, clients: make(map[uint64]*tcpClient), writeTimeout: time.Duration(opts.WriteTimeoutMS) * time.Millisecond, + tlsConfig: tlsCfg, } t.lastProcessed.Store(time.Time{}) @@ -133,17 +143,22 @@ func NewTCPSinkPlugin( "component", "tcp_sink", "instance_id", id, "host", opts.Host, - "port", opts.Port) + "port", opts.Port, + "tls", tlsCfg != nil, + "mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert) return t, nil } // Capabilities returns supported capabilities func (t *TCPSink) Capabilities() []core.Capability { - // CapTLS/CapAuth appended when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, + caps := []core.Capability{core.CapSessionAware, core.CapMultiSession} + if t.tlsConfig != nil { + caps = append(caps, core.CapTLS) + if t.tlsConfig.ClientAuth == tls.RequireAndVerifyClientCert { + caps = append(caps, core.CapAuth) // mTLS is authentication + } } + return caps } // Input returns the channel for sending transport events @@ -151,10 +166,9 @@ func (t *TCPSink) Input() chan<- core.TransportEvent { return t.input } -// listen creates the server listener. -// TLS extension point: wrap the returned listener with tls.NewListener here -// once cert config lands; no other code path changes. mTLS peer identity is -// then available via conn.(*tls.Conn).ConnectionState() in the auth hook. +// listen creates the server listener, TLS-wrapped when configured. +// Handshake is deferred: tls.NewListener conns handshake explicitly in +// handleConn under tlsx.HandshakeTimeout, post max_connections admission. func (t *TCPSink) listen() (net.Listener, error) { lc := net.ListenConfig{} if t.config.KeepAlive { @@ -164,7 +178,14 @@ func (t *TCPSink) listen() (net.Listener, error) { } } // IPv4-only, parity with existing network sinks - return lc.Listen(context.Background(), "tcp4", t.addr) + ln, err := lc.Listen(context.Background(), "tcp4", t.addr) + if err != nil { + return nil, err + } + if t.tlsConfig != nil { + ln = tls.NewListener(ln, t.tlsConfig) + } + return ln, nil } // Start binds the listener and launches accept and broadcast loops @@ -249,8 +270,8 @@ func (t *TCPSink) acceptLoop() { continue } - // Auth extension point: credential/peer verification runs here, - // pre-registration (password preamble read or TLS peer cert check) + // Password-auth extension point: preamble verification runs in + // handleConn post-handshake, pre-registration t.wg.Add(1) go t.handleConn(conn) @@ -263,11 +284,40 @@ func (t *TCPSink) handleConn(conn net.Conn) { defer t.wg.Done() remote := conn.RemoteAddr().String() - sess := t.proxy.CreateSession(remote, map[string]any{ + // Counted from accept: max_connections bounds concurrent handshakes too + count := t.activeConns.Add(1) + defer func() { + newCount := t.activeConns.Add(-1) + t.logger.Debug("msg", "TCP connection closed", + "component", "tcp_sink", + "remote_addr", remote, + "active_connections", newCount) + }() + + meta := map[string]any{ "type": "tcp_client", "remote_addr": remote, - }) + } + if tc, ok := conn.(*tls.Conn); ok { + hctx, cancel := context.WithTimeout(context.Background(), tlsx.HandshakeTimeout) + err := tc.HandshakeContext(hctx) + cancel() + if err != nil { + t.tlsHandshakeErrors.Add(1) + t.logger.Debug("msg", "TLS handshake failed", + "component", "tcp_sink", + "remote_addr", remote, + "error", err) + conn.Close() + return + } + meta["tls"] = true + if cn := tlsx.PeerCN(tc.ConnectionState()); cn != "" { + meta["tls_peer_cn"] = cn + } + } + sess := t.proxy.CreateSession(remote, meta) c := &tcpClient{ conn: conn, send: make(chan []byte, t.config.ClientBufferSize), @@ -280,7 +330,6 @@ func (t *TCPSink) handleConn(conn net.Conn) { t.clients[id] = c t.clientsMu.Unlock() - count := t.activeConns.Add(1) t.logger.Debug("msg", "TCP connection opened", "component", "tcp_sink", "remote_addr", remote, @@ -294,11 +343,6 @@ func (t *TCPSink) handleConn(conn net.Conn) { conn.Close() <-c.closed // reader has exited t.proxy.RemoveSession(sess.ID) - newCount := t.activeConns.Add(-1) - t.logger.Debug("msg", "TCP connection closed", - "component", "tcp_sink", - "remote_addr", remote, - "active_connections", newCount) }() // Reader: sink is write-only; drain and discard inbound bytes to detect @@ -385,12 +429,14 @@ func (t *TCPSink) GetStats() sink.SinkStats { StartTime: t.startTime, LastProcessed: lastProc, Details: map[string]any{ - "host": t.config.Host, - "port": t.config.Port, - "buffer_size": t.config.BufferSize, - "write_errors": t.writeErrors.Load(), - "dropped_writes": t.droppedWrites.Load(), - "rejected_conns": t.rejectedConns.Load(), + "host": t.config.Host, + "port": t.config.Port, + "buffer_size": t.config.BufferSize, + "write_errors": t.writeErrors.Load(), + "dropped_writes": t.droppedWrites.Load(), + "rejected_conns": t.rejectedConns.Load(), + "tls": t.tlsConfig != nil, + "tls_handshake_errors": t.tlsHandshakeErrors.Load(), }, } } diff --git a/internal/sink/tcpchain/tcpchain.go b/internal/sink/tcpchain/tcpchain.go index a5c0eaf..ea627ba 100644 --- a/internal/sink/tcpchain/tcpchain.go +++ b/internal/sink/tcpchain/tcpchain.go @@ -2,6 +2,7 @@ package tcpchain import ( "context" + "crypto/tls" "encoding/json" "fmt" "math/rand/v2" @@ -18,6 +19,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/sink" + "logwisp/internal/tlsx" lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" @@ -48,6 +50,7 @@ type TCPChainSink struct { node string addr string helloLine []byte + tlsConfig *tls.Config input chan core.TransportEvent logger *log.Logger @@ -122,6 +125,10 @@ func NewTCPChainSinkPlugin( if err != nil { return nil, fmt.Errorf("hello: %w", err) } + tlsCfg, err := tlsx.Client(opts.TLS, opts.Host) + if err != nil { + return nil, err + } t := &TCPChainSink{ id: id, @@ -130,6 +137,7 @@ func NewTCPChainSinkPlugin( node: node, addr: net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)), helloLine: helloLine, + tlsConfig: tlsCfg, input: make(chan core.TransportEvent, opts.BufferSize), done: make(chan struct{}), logger: logger, @@ -152,16 +160,22 @@ func NewTCPChainSinkPlugin( "component", "tcp_chain_sink", "instance_id", id, "target", t.addr, - "node", node) + "node", node, + "tls", tlsCfg != nil, + "mtls", tlsCfg != nil && len(tlsCfg.Certificates) > 0) return t, nil } // Capabilities returns supported capabilities func (t *TCPChainSink) Capabilities() []core.Capability { - // CapTLS/CapAuth added when transport security lands - return []core.Capability{ - core.CapSessionAware, + caps := []core.Capability{core.CapSessionAware} + if t.tlsConfig != nil { + caps = append(caps, core.CapTLS) + if len(t.tlsConfig.Certificates) > 0 { + caps = append(caps, core.CapAuth) // presents client identity (mTLS) + } } + return caps } // Input returns the channel for sending transport events @@ -220,6 +234,7 @@ func (t *TCPChainSink) GetStats() sink.SinkStats { Details: map[string]any{ "target": t.addr, "node": t.node, + "tls": t.tlsConfig != nil, "connected": t.connected.Load(), "reconnects": t.reconnects.Load(), "write_errors": t.writeErrors.Load(), @@ -321,18 +336,28 @@ func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool { } } -// connect performs a single dial + hello attempt +// connect performs a single dial (+ TLS handshake) + hello attempt func (t *TCPChainSink) connect(ctx context.Context) error { - d := net.Dialer{Timeout: t.dialTimeout} + nd := net.Dialer{Timeout: t.dialTimeout} if t.config.KeepAlive { - d.KeepAliveConfig = net.KeepAliveConfig{ + nd.KeepAliveConfig = net.KeepAliveConfig{ Enable: true, Idle: time.Duration(t.config.KeepAlivePeriodMS) * time.Millisecond, } } - // IPv4-only - conn, err := d.DialContext(ctx, "tcp4", t.addr) + var conn net.Conn + var err error + if t.tlsConfig != nil { + // nd.Timeout only bounds the TCP connect; tls.Dialer runs the + // handshake under ctx, so bound dial + handshake together here + dctx, cancel := context.WithTimeout(ctx, t.dialTimeout+tlsx.HandshakeTimeout) + td := tls.Dialer{NetDialer: &nd, Config: t.tlsConfig} + conn, err = td.DialContext(dctx, "tcp4", t.addr) // IPv4-only + cancel() + } else { + conn, err = nd.DialContext(ctx, "tcp4", t.addr) // IPv4-only + } if err != nil { return err } @@ -353,7 +378,8 @@ func (t *TCPChainSink) connect(ctx context.Context) error { t.logger.Info("msg", "Chain link established", "component", "tcp_chain_sink", "target", t.addr, - "node", t.node) + "node", t.node, + "tls", t.tlsConfig != nil) return nil } diff --git a/internal/source/httpchain/httpchain.go b/internal/source/httpchain/httpchain.go index bb4e059..c232161 100644 --- a/internal/source/httpchain/httpchain.go +++ b/internal/source/httpchain/httpchain.go @@ -3,6 +3,7 @@ package httpchain import ( "bufio" "context" + "crypto/tls" "errors" "fmt" "net" @@ -19,6 +20,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/source" + "logwisp/internal/tlsx" lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" @@ -49,6 +51,9 @@ type HTTPChainSource struct { server *http.Server logger *log.Logger + // TLS + tlsConfig *tls.Config + // Session cache: one session per remote host + declared node sessions map[string]string // key -> sessionID sessionsMu sync.Mutex @@ -95,6 +100,10 @@ func NewHTTPChainSourcePlugin( if opts.ReadTimeoutMS <= 0 { opts.ReadTimeoutMS = DefaultHTTPChainSourceReadTimeoutMS } + tlsCfg, err := tlsx.Server(opts.TLS) + if err != nil { + return nil, err + } s := &HTTPChainSource{ id: id, @@ -103,6 +112,7 @@ func NewHTTPChainSourcePlugin( subscribers: make([]chan core.LogEntry, 0), sessions: make(map[string]string), logger: logger, + tlsConfig: tlsCfg, } s.lastEntryTime.Store(time.Time{}) @@ -111,17 +121,22 @@ func NewHTTPChainSourcePlugin( "instance_id", id, "host", opts.Host, "port", opts.Port, - "ingest_path", opts.IngestPath) + "ingest_path", opts.IngestPath, + "tls", tlsCfg != nil, + "mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert) return s, nil } // Capabilities returns supported capabilities func (s *HTTPChainSource) Capabilities() []core.Capability { - // CapTLS/CapAuth added when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, + caps := []core.Capability{core.CapSessionAware, core.CapMultiSession} + if s.tlsConfig != nil { + caps = append(caps, core.CapTLS) + if s.tlsConfig.ClientAuth == tls.RequireAndVerifyClientCert { + caps = append(caps, core.CapAuth) // mTLS is authentication + } } + return caps } // Subscribe returns a channel for receiving log entries @@ -150,12 +165,19 @@ func (s *HTTPChainSource) Start() error { Handler: mux, ReadTimeout: time.Duration(s.config.ReadTimeoutMS) * time.Millisecond, ReadHeaderTimeout: HTTPChainReadHeaderTimeout, - // Future: TLSConfig for transport security + // TLS handshake bounded by min(ReadTimeout, ReadHeaderTimeout) + ErrorLog: tlsx.HTTPErrorLog(s.logger, "http_chain_source"), } s.startTime = time.Now() + serve := s.server.Serve + if s.tlsConfig != nil { + s.server.TLSConfig = s.tlsConfig + serve = func(l net.Listener) error { return s.server.ServeTLS(l, "", "") } + } + go func() { - if err := s.server.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + if err := serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { s.logger.Error("msg", "HTTP chain server terminated", "component", "http_chain_source", "instance_id", s.id, @@ -215,6 +237,7 @@ func (s *HTTPChainSource) GetStats() source.SourceStats { "host": s.config.Host, "port": s.config.Port, "ingest_path": s.config.IngestPath, + "tls": s.tlsConfig != nil, "total_requests": s.totalRequests.Load(), "rejected_requests": s.rejectedRequests.Load(), "parse_errors": s.parseErrors.Load(), @@ -281,14 +304,14 @@ func (s *HTTPChainSource) handleIngest(w http.ResponseWriter, r *http.Request) { for _, entry := range entries { s.publish(entry) } - s.proxy.UpdateActivity(s.sessionFor(remoteHost, connNode)) + s.proxy.UpdateActivity(s.sessionFor(remoteHost, connNode, r.TLS)) w.Header().Set(chain.HeaderAccepted, strconv.Itoa(len(entries))) w.WriteHeader(http.StatusNoContent) } // sessionFor returns the cached session for a remote+node, recreating after idle expiry -func (s *HTTPChainSource) sessionFor(remoteHost, node string) string { +func (s *HTTPChainSource) sessionFor(remoteHost, node string, cs *tls.ConnectionState) string { key := remoteHost + "|" + node s.sessionsMu.Lock() defer s.sessionsMu.Unlock() @@ -298,10 +321,17 @@ func (s *HTTPChainSource) sessionFor(remoteHost, node string) string { return id } } - sess := s.proxy.CreateSession(remoteHost, map[string]any{ + meta := map[string]any{ "type": "http_chain", "node": node, - }) + } + if cs != nil { + meta["tls"] = true + if cn := tlsx.PeerCN(*cs); cn != "" { + meta["tls_peer_cn"] = cn + } + } + sess := s.proxy.CreateSession(remoteHost, meta) s.sessions[key] = sess.ID return sess.ID } diff --git a/internal/source/tcpchain/tcpchain.go b/internal/source/tcpchain/tcpchain.go index 2b5fedc..e7cb54a 100644 --- a/internal/source/tcpchain/tcpchain.go +++ b/internal/source/tcpchain/tcpchain.go @@ -3,7 +3,7 @@ package tcpchain import ( "bufio" "context" - "encoding/json" + "crypto/tls" "errors" "fmt" "net" @@ -18,6 +18,7 @@ import ( "logwisp/internal/plugin" "logwisp/internal/session" "logwisp/internal/source" + "logwisp/internal/tlsx" lconfig "github.com/lixenwraith/config" "github.com/lixenwraith/log" @@ -45,6 +46,10 @@ type TCPChainSource struct { conns map[net.Conn]struct{} logger *log.Logger + // TLS + tlsConfig *tls.Config + tlsHandshakeErrors atomic.Uint64 + mu sync.RWMutex ctx context.Context cancel context.CancelFunc @@ -82,6 +87,10 @@ func NewTCPChainSourcePlugin( if opts.HelloTimeoutMS <= 0 { opts.HelloTimeoutMS = DefaultChainSourceHelloTimeoutMS } + tlsCfg, err := tlsx.Server(opts.TLS) + if err != nil { + return nil, err + } s := &TCPChainSource{ id: id, @@ -90,6 +99,7 @@ func NewTCPChainSourcePlugin( subscribers: make([]chan core.LogEntry, 0), conns: make(map[net.Conn]struct{}), logger: logger, + tlsConfig: tlsCfg, } s.lastEntryTime.Store(time.Time{}) @@ -97,17 +107,22 @@ func NewTCPChainSourcePlugin( "component", "tcp_chain_source", "instance_id", id, "host", opts.Host, - "port", opts.Port) + "port", opts.Port, + "tls", tlsCfg != nil, + "mtls", tlsCfg != nil && tlsCfg.ClientAuth == tls.RequireAndVerifyClientCert) return s, nil } // Capabilities returns supported capabilities func (s *TCPChainSource) Capabilities() []core.Capability { - // CapTLS/CapAuth added when transport security lands - return []core.Capability{ - core.CapSessionAware, - core.CapMultiSession, + caps := []core.Capability{core.CapSessionAware, core.CapMultiSession} + if s.tlsConfig != nil { + caps = append(caps, core.CapTLS) + if s.tlsConfig.ClientAuth == tls.RequireAndVerifyClientCert { + caps = append(caps, core.CapAuth) // mTLS is authentication + } } + return caps } // Subscribe returns a channel for receiving log entries @@ -122,11 +137,15 @@ func (s *TCPChainSource) Subscribe() <-chan core.LogEntry { // Start binds the listener and begins accepting connections func (s *TCPChainSource) Start() error { addr := net.JoinHostPort(s.config.Host, strconv.FormatInt(s.config.Port, 10)) - // IPv4-only + // IPv4-only. TLS-wrapped when configured; handshake runs explicitly in + // handleConn under tlsx.HandshakeTimeout, pre-hello. ln, err := net.Listen("tcp4", addr) if err != nil { return fmt.Errorf("listen %s: %w", addr, err) } + if s.tlsConfig != nil { + ln = tls.NewListener(ln, s.tlsConfig) + } s.listener = ln s.ctx, s.cancel = context.WithCancel(context.Background()) s.startTime = time.Now() @@ -180,12 +199,14 @@ func (s *TCPChainSource) GetStats() source.SourceStats { StartTime: s.startTime, LastEntryTime: lastEntry, Details: map[string]any{ - "host": s.config.Host, - "port": s.config.Port, - "active_connections": s.activeConns.Load(), - "rejected_conns": s.rejectedConns.Load(), - "parse_errors": s.parseErrors.Load(), - "trust_node": s.config.TrustNode, + "host": s.config.Host, + "port": s.config.Port, + "tls": s.tlsConfig != nil, + "tls_handshake_errors": s.tlsHandshakeErrors.Load(), + "active_connections": s.activeConns.Load(), + "rejected_conns": s.rejectedConns.Load(), + "parse_errors": s.parseErrors.Load(), + "trust_node": s.config.TrustNode, }, } } @@ -238,6 +259,23 @@ func (s *TCPChainSource) handleConn(conn net.Conn) { s.activeConns.Add(-1) }() + var tlsState *tls.ConnectionState + if tc, ok := conn.(*tls.Conn); ok { + hctx, cancel := context.WithTimeout(s.ctx, tlsx.HandshakeTimeout) + err := tc.HandshakeContext(hctx) + cancel() + if err != nil { + s.tlsHandshakeErrors.Add(1) + s.logger.Warn("msg", "TLS handshake failed", + "component", "tcp_chain_source", + "remote_addr", remote, + "error", err) + return // deferred cleanup closes conn + } + cs := tc.ConnectionState() + tlsState = &cs + } + scanner := bufio.NewScanner(conn) // Oversized line (> MaxLogEntryBytes) is a protocol violation; scanner is // unrecoverable after ErrTooLong, connection terminates @@ -270,10 +308,17 @@ func (s *TCPChainSource) handleConn(conn net.Conn) { } } - sess := s.proxy.CreateSession(remote, map[string]any{ + meta := map[string]any{ "type": "tcp_chain", "node": connNode, - }) + } + if tlsState != nil { + meta["tls"] = true + if cn := tlsx.PeerCN(*tlsState); cn != "" { + meta["tls_peer_cn"] = cn + } + } + sess := s.proxy.CreateSession(remote, meta) sessID = sess.ID s.logger.Info("msg", "Chain connection established", @@ -315,26 +360,6 @@ func (s *TCPChainSource) handleConn(conn net.Conn) { } } -// parseEntry decodes a canonical LogEntry line and applies the node policy -func (s *TCPChainSource) parseEntry(line []byte, connNode string) (core.LogEntry, bool) { - var entry core.LogEntry - if err := json.Unmarshal(line, &entry); err != nil { - s.parseErrors.Add(1) - s.logger.Debug("msg", "Dropped malformed chain entry", - "component", "tcp_chain_source", - "error", err) - return core.LogEntry{}, false - } - if entry.Time.IsZero() { - entry.Time = time.Now() - } - if entry.Node == "" || !s.config.TrustNode { - entry.Node = connNode - } - entry.RawSize = int64(len(line)) - return entry, true -} - // publish sends a log entry to all subscribers func (s *TCPChainSource) publish(entry core.LogEntry) { s.mu.RLock() diff --git a/internal/tlsx/tlsx.go b/internal/tlsx/tlsx.go new file mode 100644 index 0000000..8dff831 --- /dev/null +++ b/internal/tlsx/tlsx.go @@ -0,0 +1,143 @@ +// Package tlsx builds crypto/tls configurations from config.TLSOptions. +// It is the single seam between declarative TLS config and the stdlib; +// each network plugin calls exactly one constructor. +package tlsx + +import ( + "crypto/tls" + "crypto/x509" + "fmt" + stdlog "log" + "os" + "strings" + "time" + + "logwisp/internal/config" + + "github.com/lixenwraith/log" +) + +// HandshakeTimeout bounds TLS handshakes on both accept and dial paths +const HandshakeTimeout = 10 * time.Second + +// Server builds the *tls.Config for listener plugins +// (tcp/http sinks, tcp_chain/http_chain sources). Returns (nil, nil) when disabled. +func Server(o *config.TLSOptions) (*tls.Config, error) { + if o == nil || !o.Enabled { + return nil, nil + } + if o.CertFile == "" || o.KeyFile == "" { + return nil, fmt.Errorf("tls: cert_file and key_file are required for listeners") + } + cert, err := tls.LoadX509KeyPair(o.CertFile, o.KeyFile) + if err != nil { + return nil, fmt.Errorf("tls: load keypair: %w", err) + } + mv, err := minVersion(o.MinVersion) + if err != nil { + return nil, err + } + cfg := &tls.Config{ + Certificates: []tls.Certificate{cert}, + MinVersion: mv, + } + if o.ClientAuth { + if o.ClientCAFile == "" { + return nil, fmt.Errorf("tls: client_auth requires client_ca_file") + } + pool, err := loadPool(o.ClientCAFile) + if err != nil { + return nil, err + } + cfg.ClientCAs = pool + cfg.ClientAuth = tls.RequireAndVerifyClientCert + } + return cfg, nil +} + +// Client builds the *tls.Config for dialer plugins (tcp_chain/http_chain +// sinks). host seeds ServerName when no override is set; Go verifies IP SANs +// when host is an address. Returns (nil, nil) when disabled. +func Client(o *config.TLSOptions, host string) (*tls.Config, error) { + if o == nil || !o.Enabled { + return nil, nil + } + mv, err := minVersion(o.MinVersion) + if err != nil { + return nil, err + } + cfg := &tls.Config{ + MinVersion: mv, + ServerName: o.ServerName, + InsecureSkipVerify: o.InsecureSkipVerify, + } + if cfg.ServerName == "" { + cfg.ServerName = host + } + if o.CAFile != "" { + pool, err := loadPool(o.CAFile) + if err != nil { + return nil, err + } + cfg.RootCAs = pool + } + if (o.CertFile == "") != (o.KeyFile == "") { + return nil, fmt.Errorf("tls: cert_file and key_file must be set together") + } + if o.CertFile != "" { + cert, err := tls.LoadX509KeyPair(o.CertFile, o.KeyFile) + if err != nil { + return nil, fmt.Errorf("tls: load keypair: %w", err) + } + cfg.Certificates = []tls.Certificate{cert} + } + return cfg, nil +} + +// PeerCN returns the subject CN of the verified peer leaf, "" if none +func PeerCN(cs tls.ConnectionState) string { + if len(cs.PeerCertificates) == 0 { + return "" + } + return cs.PeerCertificates[0].Subject.CommonName +} + +// HTTPErrorLog adapts the structured logger for http.Server.ErrorLog so TLS +// handshake failures don't bypass log routing straight to stderr (which would +// violate the console sanitization policy). +func HTTPErrorLog(l *log.Logger, component string) *stdlog.Logger { + return stdlog.New(errLogWriter{l: l, component: component}, "", 0) +} + +type errLogWriter struct { + l *log.Logger + component string +} + +func (w errLogWriter) Write(p []byte) (int, error) { + w.l.Warn("msg", strings.TrimSpace(string(p)), "component", w.component) + return len(p), nil +} + +func loadPool(file string) (*x509.CertPool, error) { + pemBytes, err := os.ReadFile(file) + if err != nil { + return nil, fmt.Errorf("tls: read CA file: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(pemBytes) { + return nil, fmt.Errorf("tls: no certificates found in %s", file) + } + return pool, nil +} + +func minVersion(s string) (uint16, error) { + switch s { + case "", "1.3": + return tls.VersionTLS13, nil + case "1.2": + return tls.VersionTLS12, nil + default: + return 0, fmt.Errorf("tls: min_version %q (valid: \"1.2\", \"1.3\")", s) + } +}