From c4fa90fde4da9d85af745a5a7dfd832ee9e9143f Mon Sep 17 00:00:00 2001 From: bubunyo nyavor Date: Thu, 11 Jun 2026 09:54:57 +0200 Subject: [PATCH 1/2] add tls termination --- README.md | 16 +++- cmd/kroxy/main.go | 6 ++ config/config.go | 32 +++++++ config/config_test.go | 97 +++++++++++++++++++++ dockerfiles/kroxy.yaml | 7 ++ proxy/listener.go | 9 +- proxy/listener_tls_test.go | 171 +++++++++++++++++++++++++++++++++++++ 7 files changed, 335 insertions(+), 3 deletions(-) create mode 100644 proxy/listener_tls_test.go diff --git a/README.md b/README.md index 281e908..e202279 100644 --- a/README.md +++ b/README.md @@ -143,6 +143,11 @@ kroxy reads a single YAML file. Minimal example: listen: ":9092" # client-facing Kafka listener advertised: "kroxy:9092" # what kroxy advertises as broker 0 +tls: # optional; omit for a plaintext listener + enabled: true + cert_file: /etc/kroxy/certs/server.crt + key_file: /etc/kroxy/certs/server.key + upstream: bootstrap: "kafka:9093" # default upstream for tenants that omit it @@ -178,6 +183,11 @@ Notes: - The `resolver.memory.tenants` list may be empty **only if** the admin RPC is enabled — otherwise the proxy has nothing to authorise against. - A tenant's `id` and `topic_prefix` are both required. +- `tls` is optional. When `enabled`, kroxy terminates TLS on the client + listener using `cert_file`/`key_file` (both required); the upstream broker + connection is unaffected. Omit the block for a plaintext listener. Server-side + TLS only — clients are not asked for a certificate, and the keypair is loaded + once at startup (restart to rotate). ## Authentication model @@ -289,8 +299,10 @@ examples/ # admin-curl.sh helper v1 is deliberately small. The following are explicitly out of scope and deferred: -- **No TLS** on either the client or upstream side. Run kroxy on a - trusted network or behind a TLS-terminating sidecar. +- **Client TLS termination** is supported via the `tls` config block + (server-side only, no mTLS, no hot reload). **Upstream TLS is not** — the + connection to the broker is always plaintext, so run kroxy on a trusted + network relative to the broker. - **SASL/PLAIN, SCRAM-SHA-256, SCRAM-SHA-512.** No OAUTHBEARER, no mTLS, no Kerberos. No SASL channel binding. - **Single shared upstream cluster.** Per-tenant `upstream` is plumbed diff --git a/cmd/kroxy/main.go b/cmd/kroxy/main.go index 132c7dc..6195388 100644 --- a/cmd/kroxy/main.go +++ b/cmd/kroxy/main.go @@ -58,9 +58,15 @@ func run() error { metrics = observability.NewMetrics() } + tlsCfg, err := cfg.TLS.Build() + if err != nil { + return err + } + srv := proxy.NewServer(proxy.ServerConfig{ Listen: cfg.Listen, Advertised: cfg.Advertised, + TLS: tlsCfg, }, res, metrics, log) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) diff --git a/config/config.go b/config/config.go index 4596417..c230c17 100644 --- a/config/config.go +++ b/config/config.go @@ -2,6 +2,7 @@ package config import ( + "crypto/tls" "net" "os" @@ -14,6 +15,7 @@ import ( type Config struct { Listen string `yaml:"listen"` Advertised string `yaml:"advertised"` + TLS TLSConfig `yaml:"tls"` Upstream UpstreamConfig `yaml:"upstream"` Resolver resolver.Config `yaml:"resolver"` Log LogConfig `yaml:"log"` @@ -21,6 +23,33 @@ type Config struct { Admin AdminConfig `yaml:"admin"` } +// TLSConfig configures TLS termination on the client-facing listener. When +// disabled (the default) the listener is plaintext. kroxy is the TLS endpoint; +// the upstream broker connection is unaffected. Server-side TLS only — clients +// are not asked for a certificate. +type TLSConfig struct { + Enabled bool `yaml:"enabled"` + CertFile string `yaml:"cert_file"` + KeyFile string `yaml:"key_file"` +} + +// Build loads the keypair and returns the listener's *tls.Config, or nil when +// TLS is disabled (so callers can pass the result straight through and treat +// nil as plaintext). +func (t TLSConfig) Build() (*tls.Config, error) { + if !t.Enabled { + return nil, nil + } + cert, err := tls.LoadX509KeyPair(t.CertFile, t.KeyFile) + if err != nil { + return nil, errors.Wrap(err, "TLSConfig.Build") + } + return &tls.Config{ + Certificates: []tls.Certificate{cert}, + MinVersion: tls.VersionTLS12, + }, nil +} + // MetricsConfig configures the Prometheus metrics endpoint. type MetricsConfig struct { Enabled bool `yaml:"enabled"` @@ -111,6 +140,9 @@ func (c *Config) validate() error { return errors.Wrapf(err, "config: admin.listen is invalid") } } + if c.TLS.Enabled && (c.TLS.CertFile == "" || c.TLS.KeyFile == "") { + return errors.New("config: tls.cert_file and tls.key_file are required when tls.enabled") + } return nil } diff --git a/config/config_test.go b/config/config_test.go index ca29edd..c48f9d4 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -1,9 +1,18 @@ package config_test import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" "os" "path/filepath" "testing" + "time" "github.com/bubunyo/kroxy/config" "github.com/stretchr/testify/assert" @@ -18,6 +27,58 @@ func writeFile(t *testing.T, contents string) string { return p } +// writeKeyPair writes a self-signed cert/key PEM pair into a temp dir and +// returns their paths. +func writeKeyPair(t *testing.T) (certPath, keyPath string) { + t.Helper() + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "kroxy-test"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &priv.PublicKey, priv) + require.NoError(t, err) + keyDER, err := x509.MarshalPKCS8PrivateKey(priv) + require.NoError(t, err) + + dir := t.TempDir() + certPath = filepath.Join(dir, "server.crt") + keyPath = filepath.Join(dir, "server.key") + require.NoError(t, os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600)) + require.NoError(t, os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: keyDER}), 0o600)) + return certPath, keyPath +} + +func TestTLSConfig_Build(t *testing.T) { + t.Parallel() + + t.Run("disabled returns nil", func(t *testing.T) { + t.Parallel() + got, err := config.TLSConfig{Enabled: false}.Build() + require.NoError(t, err) + assert.Nil(t, got) + }) + + t.Run("enabled loads keypair", func(t *testing.T) { + t.Parallel() + cert, key := writeKeyPair(t) + got, err := config.TLSConfig{Enabled: true, CertFile: cert, KeyFile: key}.Build() + require.NoError(t, err) + require.NotNil(t, got) + assert.Len(t, got.Certificates, 1) + assert.Equal(t, uint16(tls.VersionTLS12), got.MinVersion) + }) + + t.Run("enabled with missing file errors", func(t *testing.T) { + t.Parallel() + _, err := config.TLSConfig{Enabled: true, CertFile: "/nope.crt", KeyFile: "/nope.key"}.Build() + require.Error(t, err) + }) +} + func TestLoad(t *testing.T) { t.Parallel() @@ -144,6 +205,42 @@ resolver: `, wantErr: true, }, + { + name: "tls enabled without cert/key", + yaml: ` +advertised: "kroxy:9092" +upstream: { bootstrap: "k:9092" } +resolver: + memory: + tenants: + - id: tenantA + topic_prefix: "tenantA." +tls: + enabled: true +`, + wantErr: true, + }, + { + name: "tls enabled with cert and key", + yaml: ` +advertised: "kroxy:9092" +upstream: { bootstrap: "k:9092" } +resolver: + memory: + tenants: + - id: tenantA + topic_prefix: "tenantA." +tls: + enabled: true + cert_file: /etc/kroxy/certs/server.crt + key_file: /etc/kroxy/certs/server.key +`, + check: func(t *testing.T, c config.Config) { + assert.True(t, c.TLS.Enabled) + assert.Equal(t, "/etc/kroxy/certs/server.crt", c.TLS.CertFile) + assert.Equal(t, "/etc/kroxy/certs/server.key", c.TLS.KeyFile) + }, + }, } for _, tt := range tests { diff --git a/dockerfiles/kroxy.yaml b/dockerfiles/kroxy.yaml index 297e39e..81d6b84 100644 --- a/dockerfiles/kroxy.yaml +++ b/dockerfiles/kroxy.yaml @@ -1,6 +1,13 @@ listen: ":9092" advertised: "kroxy:9092" +# Terminate TLS on the client-facing listener. Disabled here so the demo stack +# stays plaintext; mount a cert/key and enable to serve clients over TLS. +# tls: +# enabled: true +# cert_file: /etc/kroxy/certs/server.crt +# key_file: /etc/kroxy/certs/server.key + upstream: bootstrap: "kafka:9093" diff --git a/proxy/listener.go b/proxy/listener.go index 7cfafba..ed7c791 100644 --- a/proxy/listener.go +++ b/proxy/listener.go @@ -5,6 +5,7 @@ package proxy import ( "context" + "crypto/tls" "errors" "log/slog" "net" @@ -31,6 +32,9 @@ type Server struct { type ServerConfig struct { Listen string Advertised string + // TLS, when non-nil, terminates TLS on the client-facing listener. A nil + // value leaves the listener plaintext. + TLS *tls.Config } // NewServer constructs a Server. It does not start listening; call Run. @@ -47,8 +51,11 @@ func (s *Server) Run(ctx context.Context) error { if err != nil { return pkgerrors.Wrap(err, "Run") } + if s.cfg.TLS != nil { + ln = tls.NewListener(ln, s.cfg.TLS) + } s.listener = ln - s.log.InfoContext(ctx, "kroxy listening", "addr", ln.Addr().String(), "advertised", s.cfg.Advertised) + s.log.InfoContext(ctx, "kroxy listening", "addr", ln.Addr().String(), "advertised", s.cfg.Advertised, "tls", s.cfg.TLS != nil) go func() { <-ctx.Done() diff --git a/proxy/listener_tls_test.go b/proxy/listener_tls_test.go new file mode 100644 index 0000000..730634d --- /dev/null +++ b/proxy/listener_tls_test.go @@ -0,0 +1,171 @@ +package proxy_test + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "io" + "log/slog" + "math/big" + "net" + "testing" + "time" + + "github.com/bubunyo/kroxy/protocol" + "github.com/bubunyo/kroxy/proxy" + "github.com/bubunyo/kroxy/resolver" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kmsg" +) + +// genSelfSignedCert returns a self-signed server certificate valid for the +// loopback addresses, plus a CertPool that trusts it. +func genSelfSignedCert(t *testing.T) (tls.Certificate, *x509.CertPool) { + t.Helper() + priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "kroxy-test"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: true, + DNSNames: []string{"localhost"}, + IPAddresses: []net.IP{net.ParseIP("127.0.0.1"), net.IPv6loopback}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &priv.PublicKey, priv) + require.NoError(t, err) + + leaf, err := x509.ParseCertificate(der) + require.NoError(t, err) + pool := x509.NewCertPool() + pool.AddCert(leaf) + + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: priv, Leaf: leaf}, pool +} + +// startTestServerTLS starts a TLS-terminating proxy and returns its address, +// a CertPool trusting its certificate, and a stop func. +func startTestServerTLS(t *testing.T) (string, *x509.CertPool, func()) { + t.Helper() + r, err := resolver.New(resolver.Config{ + Memory: resolver.MemoryConfig{Tenants: []resolver.Tenant{ + {ID: "alice", TopicPrefix: "tenantA.", Upstream: "kafka:9092"}, + }}, + }) + require.NoError(t, err) + + cert, pool := genSelfSignedCert(t) + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := ln.Addr().String() + require.NoError(t, ln.Close()) + + srv := proxy.NewServer(proxy.ServerConfig{ + Listen: addr, + Advertised: addr, + TLS: &tls.Config{Certificates: []tls.Certificate{cert}, MinVersion: tls.VersionTLS12}, + }, r, nil, log) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + _ = srv.Run(ctx) + close(done) + }() + + deadline := time.Now().Add(2 * time.Second) + for { + c, err := net.Dial("tcp", addr) + if err == nil { + _ = c.Close() + break + } + if time.Now().After(deadline) { + cancel() + t.Fatalf("server never came up: %v", err) + } + time.Sleep(10 * time.Millisecond) + } + + return addr, pool, func() { cancel(); <-done } +} + +// TestTLS_HandshakeThenSaslPlain verifies a TLS client completes the SASL/PLAIN +// flow through a TLS-terminating listener. SASL runs inside the TLS session. +func TestTLS_HandshakeThenSaslPlain(t *testing.T) { + t.Parallel() + + addr, pool, stop := startTestServerTLS(t) + defer stop() + + c, err := tls.Dial("tcp", addr, &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}) + require.NoError(t, err) + defer c.Close() + require.NoError(t, c.SetDeadline(time.Now().Add(3*time.Second))) + + avReq := kmsg.NewPtrApiVersionsRequest() + avReq.SetVersion(3) + avReq.ClientSoftwareName = "test" + avReq.ClientSoftwareVersion = "0" + sendRequest(t, c, avReq, 1, "test") + avResp := kmsg.NewPtrApiVersionsResponse() + avResp.SetVersion(3) + cid := recvResponse(t, c, avResp, protocol.ApiVersionsKey, 3) + assert.Equal(t, int32(1), cid) + assert.Equal(t, int16(0), avResp.ErrorCode) + + hsReq := kmsg.NewPtrSASLHandshakeRequest() + hsReq.SetVersion(1) + hsReq.Mechanism = "PLAIN" + sendRequest(t, c, hsReq, 2, "test") + hsResp := kmsg.NewPtrSASLHandshakeResponse() + hsResp.SetVersion(1) + _ = recvResponse(t, c, hsResp, protocol.SaslHandshakeKey, 1) + assert.Equal(t, int16(0), hsResp.ErrorCode) + + authReq := kmsg.NewPtrSASLAuthenticateRequest() + authReq.SetVersion(1) + authReq.SASLAuthBytes = []byte("\x00alice\x00alicepw") + sendRequest(t, c, authReq, 3, "test") + authResp := kmsg.NewPtrSASLAuthenticateResponse() + authResp.SetVersion(1) + _ = recvResponse(t, c, authResp, protocol.SaslAuthenticateKey, 1) + assert.Equal(t, int16(0), authResp.ErrorCode, "expected auth ok, got msg=%v", strOrNil(authResp.ErrorMessage)) +} + +// TestTLS_PlaintextClientRejected verifies a plaintext client cannot speak to a +// TLS listener: the Kafka request bytes are not a valid TLS ClientHello, so the +// server aborts the handshake and the client gets no valid frame back. +func TestTLS_PlaintextClientRejected(t *testing.T) { + t.Parallel() + + addr, _, stop := startTestServerTLS(t) + defer stop() + + c, err := net.Dial("tcp", addr) + require.NoError(t, err) + defer c.Close() + require.NoError(t, c.SetDeadline(time.Now().Add(3*time.Second))) + + avReq := kmsg.NewPtrApiVersionsRequest() + avReq.SetVersion(3) + avReq.ClientSoftwareName = "test" + avReq.ClientSoftwareVersion = "0" + f := kmsg.NewRequestFormatter(kmsg.FormatterClientID("test")) + _, _ = c.Write(f.AppendRequest(nil, avReq, 1)) + + _, err = protocol.ReadFrame(c) + require.Error(t, err, "plaintext client must not receive a valid frame from a TLS listener") +} From e5b6238915133b302af7efe02bd0150759c792ff Mon Sep 17 00:00:00 2001 From: bubunyo nyavor Date: Thu, 11 Jun 2026 10:05:33 +0200 Subject: [PATCH 2/2] add back off and handle pr comment --- proxy/listener.go | 40 +++++++++++++++++++++++++++++++------- proxy/listener_tls_test.go | 34 ++++++++++++++++++++++++++++++++ 2 files changed, 67 insertions(+), 7 deletions(-) diff --git a/proxy/listener.go b/proxy/listener.go index ed7c791..4e99fc9 100644 --- a/proxy/listener.go +++ b/proxy/listener.go @@ -43,8 +43,18 @@ func NewServer(cfg ServerConfig, r resolver.Resolver, m *observability.Metrics, return &Server{cfg: cfg, resolver: r, metrics: m, log: log} } -// Run begins accepting connections until ctx is cancelled or the listener -// returns a non-temporary error. It blocks the caller. +// maxAcceptBackoff caps the retry delay applied after a transient Accept error. +const maxAcceptBackoff = time.Second + +// Run begins accepting connections until ctx is cancelled or the listener is +// closed. It blocks the caller. +// +// Transient Accept errors (fd exhaustion, a connection reset between accept and +// return, etc.) are logged and retried with a capped exponential backoff rather +// than treated as fatal — a single misbehaving client must never tear down the +// proxy. When TLS is enabled the handshake is deferred to the first read, so a +// failed handshake surfaces per-connection in handle (logged, non-fatal), not +// here. func (s *Server) Run(ctx context.Context) error { lc := net.ListenConfig{} ln, err := lc.Listen(ctx, "tcp", s.cfg.Listen) @@ -62,19 +72,35 @@ func (s *Server) Run(ctx context.Context) error { _ = ln.Close() }() + var backoff time.Duration for { c, err := ln.Accept() if err != nil { - if ctx.Err() != nil { + // Clean shutdown: ctx cancelled or the listener was closed. + if ctx.Err() != nil || errors.Is(err, net.ErrClosed) { s.wg.Wait() return nil } - var ne net.Error - if errors.As(err, &ne) && ne.Timeout() { - continue + if backoff == 0 { + backoff = 5 * time.Millisecond + } else { + backoff *= 2 + } + if backoff > maxAcceptBackoff { + backoff = maxAcceptBackoff + } + s.log.WarnContext(ctx, "accept error; retrying", "err", err, "delay", backoff.String()) + t := time.NewTimer(backoff) + select { + case <-ctx.Done(): + t.Stop() + s.wg.Wait() + return nil + case <-t.C: } - return pkgerrors.Wrap(err, "Run") + continue } + backoff = 0 s.wg.Go(func() { s.handle(ctx, c) }) } } diff --git a/proxy/listener_tls_test.go b/proxy/listener_tls_test.go index 730634d..a4fe242 100644 --- a/proxy/listener_tls_test.go +++ b/proxy/listener_tls_test.go @@ -169,3 +169,37 @@ func TestTLS_PlaintextClientRejected(t *testing.T) { _, err = protocol.ReadFrame(c) require.Error(t, err, "plaintext client must not receive a valid frame from a TLS listener") } + +// TestTLS_BadClientDoesNotKillServer asserts a failed/garbage connection does +// not tear down the listener: a subsequent valid TLS client still completes the +// SASL flow. +func TestTLS_BadClientDoesNotKillServer(t *testing.T) { + t.Parallel() + + addr, pool, stop := startTestServerTLS(t) + defer stop() + + // A bad client: plaintext garbage against the TLS listener. Its handshake + // fails in the per-connection handler, not the accept loop. + bad, err := net.Dial("tcp", addr) + require.NoError(t, err) + _, _ = bad.Write([]byte("not a tls client hello")) + _ = bad.Close() + + // The server must still serve a well-behaved TLS client. + good, err := tls.Dial("tcp", addr, &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}) + require.NoError(t, err) + defer good.Close() + require.NoError(t, good.SetDeadline(time.Now().Add(3*time.Second))) + + avReq := kmsg.NewPtrApiVersionsRequest() + avReq.SetVersion(3) + avReq.ClientSoftwareName = "test" + avReq.ClientSoftwareVersion = "0" + sendRequest(t, good, avReq, 1, "test") + avResp := kmsg.NewPtrApiVersionsResponse() + avResp.SetVersion(3) + cid := recvResponse(t, good, avResp, protocol.ApiVersionsKey, 3) + assert.Equal(t, int32(1), cid) + assert.Equal(t, int16(0), avResp.ErrorCode) +}