add tls termination

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

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)

config/config.go

@@ -2,6 +2,7 @@
 package config
 
 import (
+	"crypto/tls"
 	"net"
 	"os"
 
@@ -14,13 +15,41 @@ 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"`
 	Metrics    MetricsConfig   `yaml:"metrics"`
 	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
 }
 

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 {

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"
 

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.
@@ -39,35 +43,64 @@ 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)
 	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()
 		_ = 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) })
 	}
 }