Feat/enable sasl types

README.md

@@ -150,6 +150,8 @@ tls:                          # optional; omit for a plaintext listener
 
 upstream:
   bootstrap: "kafka:9093"     # default upstream for tenants that omit it
+  sasl:                       # optional; omit for PLAIN pass-through
+    mechanism: SCRAM-SHA-256  # authenticate PLAIN clients to the broker via SCRAM
 
 resolver:
   type: memory                # only "memory" is supported in v1
@@ -188,6 +190,11 @@ Notes:
   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).
+- `upstream.sasl.mechanism` is optional. Leave it empty (the default) to
+  forward PLAIN clients' credentials to the broker verbatim. Set it to
+  `SCRAM-SHA-256` or `SCRAM-SHA-512` to make kroxy authenticate PLAIN clients
+  to the upstream broker with that SCRAM mechanism instead — see
+  [PLAIN→SCRAM translation](#plainscram-translation).
 
 ## Authentication model
 
@@ -222,6 +229,22 @@ Consequences:
 The only thing kroxy needs to know about a tenant is the mapping
 `id → (topic_prefix, upstream)`.
 
+### PLAIN→SCRAM translation
+
+Some clients can only speak SASL/PLAIN, while the broker provisions
+per-tenant **SCRAM** credentials and no static PLAIN principals. Setting
+`upstream.sasl.mechanism` to a SCRAM mechanism bridges the two: a client
+authenticates to kroxy with PLAIN, and kroxy authenticates to the upstream
+broker with SCRAM, computing the proof itself from the tenant ID and the
+client-supplied password. The tenant's SCRAM password must therefore equal
+the PLAIN password the client sends.
+
+This is distinct from the pass-through SCRAM mechanism above. It applies
+**only** to clients that connect with PLAIN — kroxy needs the plaintext
+password to compute a SCRAM proof, and only PLAIN reveals it. Clients that
+already speak SCRAM are relayed verbatim regardless of this setting, and when
+`upstream.sasl.mechanism` is empty kroxy holds no secret and changes nothing.
+
 ## Admin RPC
 
 kroxy exposes a JSON-RPC 2.0 admin API for managing tenants at runtime. See [admin/RPC.md](admin/RPC.md) for methods, request/response shapes, error codes, and curl examples.

cmd/kroxy/main.go

@@ -64,9 +64,10 @@ func run() error {
 	}
 
 	srv := proxy.NewServer(proxy.ServerConfig{
-		Listen:     cfg.Listen,
-		Advertised: cfg.Advertised,
-		TLS:        tlsCfg,
+		Listen:                cfg.Listen,
+		Advertised:            cfg.Advertised,
+		TLS:                   tlsCfg,
+		UpstreamSASLMechanism: cfg.Upstream.SASL.Mechanism,
 	}, res, metrics, log)
 
 	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)

config/config.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"os"
 
+	"github.com/bubunyo/kroxy/auth"
 	"github.com/bubunyo/kroxy/resolver"
 	"github.com/pkg/errors"
 	"gopkg.in/yaml.v3"
@@ -67,7 +68,23 @@ type AdminConfig struct {
 
 // UpstreamConfig describes the shared upstream Kafka cluster.
 type UpstreamConfig struct {
-	Bootstrap string `yaml:"bootstrap"`
+	Bootstrap string             `yaml:"bootstrap"`
+	SASL      UpstreamSASLConfig `yaml:"sasl"`
+}
+
+// UpstreamSASLConfig controls how kroxy authenticates to the upstream broker
+// for clients that connect with SASL/PLAIN.
+//
+// When Mechanism is empty (the default), kroxy forwards the client's PLAIN
+// credentials to the broker verbatim — pure pass-through. When Mechanism is a
+// SCRAM mechanism, kroxy instead authenticates upstream with that mechanism,
+// computing the SCRAM proof itself from the tenant ID and the client-supplied
+// PLAIN password. This lets PLAIN-only clients reach a broker that only
+// provisions per-tenant SCRAM credentials, without kroxy holding any static
+// secret. It has no effect on clients that already speak SCRAM (those are
+// relayed verbatim).
+type UpstreamSASLConfig struct {
+	Mechanism string `yaml:"mechanism"`
 }
 
 // LogConfig configures the slog handler.
@@ -143,6 +160,9 @@ func (c *Config) validate() error {
 	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")
 	}
+	if m := c.Upstream.SASL.Mechanism; m != "" && !auth.IsSCRAMMechanism(m) {
+		return errors.Errorf("config: upstream.sasl.mechanism %q is not supported (use SCRAM-SHA-256 or SCRAM-SHA-512, or leave empty for PLAIN pass-through)", m)
+	}
 	return nil
 }
 

config/config_test.go

@@ -241,6 +241,71 @@ tls:
 				assert.Equal(t, "/etc/kroxy/certs/server.key", c.TLS.KeyFile)
 			},
 		},
+		{
+			name: "upstream sasl scram-256 valid",
+			yaml: `
+advertised: "kroxy:9092"
+upstream:
+  bootstrap: "k:9092"
+  sasl:
+    mechanism: SCRAM-SHA-256
+resolver:
+  memory:
+    tenants:
+      - id: tenantA
+        topic_prefix: "tenantA."
+`,
+			check: func(t *testing.T, c config.Config) {
+				assert.Equal(t, "SCRAM-SHA-256", c.Upstream.SASL.Mechanism)
+			},
+		},
+		{
+			name: "upstream sasl empty is plain passthrough",
+			yaml: `
+advertised: "kroxy:9092"
+upstream: { bootstrap: "k:9092" }
+resolver:
+  memory:
+    tenants:
+      - id: tenantA
+        topic_prefix: "tenantA."
+`,
+			check: func(t *testing.T, c config.Config) {
+				assert.Empty(t, c.Upstream.SASL.Mechanism)
+			},
+		},
+		{
+			name: "upstream sasl plain rejected",
+			yaml: `
+advertised: "kroxy:9092"
+upstream:
+  bootstrap: "k:9092"
+  sasl:
+    mechanism: PLAIN
+resolver:
+  memory:
+    tenants:
+      - id: tenantA
+        topic_prefix: "tenantA."
+`,
+			wantErr: true,
+		},
+		{
+			name: "upstream sasl unknown mechanism rejected",
+			yaml: `
+advertised: "kroxy:9092"
+upstream:
+  bootstrap: "k:9092"
+  sasl:
+    mechanism: GSSAPI
+resolver:
+  memory:
+    tenants:
+      - id: tenantA
+        topic_prefix: "tenantA."
+`,
+			wantErr: true,
+		},
 	}
 
 	for _, tt := range tests {

dockerfiles/kroxy.yaml

@@ -10,6 +10,10 @@ advertised: "kroxy:9092"
 
 upstream:
   bootstrap: "kafka:9093"
+  # Authenticate PLAIN clients to the broker via SCRAM instead of forwarding
+  # PLAIN verbatim. Omitted here so the demo stays PLAIN pass-through.
+  # sasl:
+  #   mechanism: SCRAM-SHA-256
 
 resolver:
   type: memory

proxy/conn.go

@@ -45,7 +45,7 @@ type conn struct {
 	log      *slog.Logger
 
 	state                connState
-	mechanism            string
+	clientMechanism      string // SASL mechanism the client used to authenticate to kroxy
 	tenant               resolver.Tenant
 	password             string // populated only on the PLAIN path
 	upstream             *upstream.Conn
@@ -201,19 +201,41 @@ func (c *conn) forwardRewritten(hdr protocol.RequestHeader, body []byte) error {
 	return h(hdr, body)
 }
 
+// shouldTranslatePlainToSCRAM reports whether this connection's upstream dial
+// must bridge a PLAIN client to a SCRAM broker. It is true only when the
+// client authenticated to kroxy with PLAIN (so kroxy holds the plaintext
+// password needed to compute a SCRAM proof) and kroxy is configured with an
+// upstream SCRAM mechanism. Clients that already speak SCRAM authenticate
+// during handshake via the relay path and never reach ensureUpstream.
+func (c *conn) shouldTranslatePlainToSCRAM() bool {
+	return c.clientMechanism == auth.MechanismPlain && auth.IsSCRAMMechanism(c.cfg.UpstreamSASLMechanism)
+}
+
 func (c *conn) ensureUpstream() error {
 	if c.upstream != nil {
 		return nil
 	}
-	up, err := upstream.Dial(c.ctx, c.tenant.Upstream, c.tenant.ID, c.password)
+	var (
+		up  *upstream.Conn
+		err error
+	)
+	if c.shouldTranslatePlainToSCRAM() {
+		up, err = upstream.DialSCRAM(c.ctx, c.tenant.Upstream, c.cfg.UpstreamSASLMechanism, c.tenant.ID, c.password)
+	} else {
+		up, err = upstream.Dial(c.ctx, c.tenant.Upstream, c.tenant.ID, c.password)
+	}
 	if err != nil {
 		if c.metrics != nil {
 			c.metrics.UpstreamErrorTotal.WithLabelValues("dial").Inc()
 		}
 		return errors.Wrap(err, "ensureUpstream")
 	}
 	c.upstream = up
-	c.log.InfoContext(c.ctx, "upstream connected", "addr", c.tenant.Upstream, "tenant_id", c.tenant.ID)
+	upstreamMech := c.clientMechanism
+	if c.shouldTranslatePlainToSCRAM() {
+		upstreamMech = c.cfg.UpstreamSASLMechanism
+	}
+	c.log.InfoContext(c.ctx, "upstream connected", "addr", c.tenant.Upstream, "tenant_id", c.tenant.ID, "upstream_mechanism", upstreamMech)
 	return nil
 }
 
@@ -276,7 +298,7 @@ func (c *conn) handleSaslHandshake(hdr protocol.RequestHeader, body []byte) erro
 		resp.ErrorCode = errUnsupportedSaslMech
 		c.observeHandshake(req.Mechanism, "unsupported")
 	default:
-		c.mechanism = req.Mechanism
+		c.clientMechanism = req.Mechanism
 		c.state = stateAwaitAuth
 	}
 	return c.writeResponse(hdr, resp)
@@ -304,7 +326,7 @@ func (c *conn) observeHandshake(mech, result string) {
 }
 
 func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) error {
-	if auth.IsSCRAMMechanism(c.mechanism) {
+	if auth.IsSCRAMMechanism(c.clientMechanism) {
 		return c.handleSaslAuthenticateSCRAM(hdr, body)
 	}
 	return c.handleSaslAuthenticatePlain(hdr, body)
@@ -383,7 +405,7 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 		resp.ErrorCode = errIllegalSaslState
 		msg := "SASL handshake required"
 		resp.ErrorMessage = &msg
-		c.observeHandshake(c.mechanism, "illegal_state")
+		c.observeHandshake(c.clientMechanism, "illegal_state")
 		return c.writeResponse(hdr, resp)
 	}
 
@@ -394,7 +416,7 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 			resp.ErrorCode = errSaslAuthFailed
 			msg := "malformed SCRAM client-first-message"
 			resp.ErrorMessage = &msg
-			c.observeHandshake(c.mechanism, "malformed")
+			c.observeHandshake(c.clientMechanism, "malformed")
 			c.log.InfoContext(c.ctx, "sasl scram parse failed", "err", err)
 			return c.writeResponse(hdr, resp)
 		}
@@ -406,7 +428,7 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 			resp.ErrorCode = errSaslAuthFailed
 			msg := "authentication failed"
 			resp.ErrorMessage = &msg
-			c.observeHandshake(c.mechanism, "unauthorized")
+			c.observeHandshake(c.clientMechanism, "unauthorized")
 			c.log.InfoContext(c.ctx, "sasl auth failed", "tenant_id", username, "err", err)
 			return c.writeResponse(hdr, resp)
 		}
@@ -415,15 +437,15 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 		}
 		c.tenant = tenant
 
-		up, dErr := upstream.DialForSCRAM(c.ctx, tenant.Upstream, c.mechanism)
+		up, dErr := upstream.DialForSCRAM(c.ctx, tenant.Upstream, c.clientMechanism)
 		if dErr != nil {
 			if c.metrics != nil {
 				c.metrics.UpstreamErrorTotal.WithLabelValues("scram_dial").Inc()
 			}
 			resp.ErrorCode = errSaslAuthFailed
 			msg := "upstream unavailable"
 			resp.ErrorMessage = &msg
-			c.observeHandshake(c.mechanism, "upstream_error")
+			c.observeHandshake(c.clientMechanism, "upstream_error")
 			c.log.WarnContext(c.ctx, "scram upstream dial failed", "tenant_id", tenant.ID, "err", dErr)
 			return c.writeResponse(hdr, resp)
 		}
@@ -436,7 +458,7 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 		if c.metrics != nil {
 			c.metrics.UpstreamErrorTotal.WithLabelValues("scram_relay").Inc()
 		}
-		c.observeHandshake(c.mechanism, "upstream_error")
+		c.observeHandshake(c.clientMechanism, "upstream_error")
 		return errors.Wrap(rErr, "handleSaslAuthenticateSCRAM")
 	}
 	resp.SASLAuthBytes = respBytes
@@ -446,16 +468,16 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by
 		resp.ErrorMessage = &m
 	}
 	if errCode != 0 {
-		c.observeHandshake(c.mechanism, "unauthorized")
+		c.observeHandshake(c.clientMechanism, "unauthorized")
 		c.log.InfoContext(c.ctx, "scram upstream rejected", "tenant_id", c.tenant.ID, "code", errCode, "msg", errMsg)
 		return c.writeResponse(hdr, resp)
 	}
 
 	c.scramRoundsCompleted++
 	if c.scramRoundsCompleted >= 2 {
 		c.state = stateAuthenticated
-		c.observeHandshake(c.mechanism, "ok")
-		c.log.InfoContext(c.ctx, "sasl auth ok", "tenant_id", c.tenant.ID, "mechanism", c.mechanism)
+		c.observeHandshake(c.clientMechanism, "ok")
+		c.log.InfoContext(c.ctx, "sasl auth ok", "tenant_id", c.tenant.ID, "mechanism", c.clientMechanism)
 	}
 	return c.writeResponse(hdr, resp)
 }

proxy/listener.go

@@ -35,6 +35,10 @@ type ServerConfig struct {
 	// TLS, when non-nil, terminates TLS on the client-facing listener. A nil
 	// value leaves the listener plaintext.
 	TLS *tls.Config
+	// UpstreamSASLMechanism, when set to a SCRAM mechanism, makes kroxy
+	// authenticate PLAIN clients to the upstream broker with that mechanism
+	// (PLAIN->SCRAM translation). Empty means forward PLAIN verbatim.
+	UpstreamSASLMechanism string
 }
 
 // NewServer constructs a Server. It does not start listening; call Run.

proxy/translate_internal_test.go

@@ -0,0 +1,34 @@
+package proxy
+
+import (
+	"testing"
+
+	"github.com/bubunyo/kroxy/auth"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestShouldTranslatePlainToSCRAM(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name          string
+		clientMech    string
+		upstreamMech  string
+		wantTranslate bool
+	}{
+		{name: "plain client, scram-256 upstream", clientMech: auth.MechanismPlain, upstreamMech: auth.MechanismSCRAMSHA256, wantTranslate: true},
+		{name: "plain client, scram-512 upstream", clientMech: auth.MechanismPlain, upstreamMech: auth.MechanismSCRAMSHA512, wantTranslate: true},
+		{name: "plain client, no upstream mech (passthrough)", clientMech: auth.MechanismPlain, upstreamMech: "", wantTranslate: false},
+		{name: "scram client never translates", clientMech: auth.MechanismSCRAMSHA256, upstreamMech: auth.MechanismSCRAMSHA256, wantTranslate: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			c := &conn{
+				clientMechanism: tt.clientMech,
+				cfg:             ServerConfig{UpstreamSASLMechanism: tt.upstreamMech},
+			}
+			assert.Equal(t, tt.wantTranslate, c.shouldTranslatePlainToSCRAM())
+		})
+	}
+}