From 7a1ea07efdf8ae2885c5eefebbc79114d402184b Mon Sep 17 00:00:00 2001 From: bubunyo nyavor Date: Thu, 11 Jun 2026 13:55:29 +0200 Subject: [PATCH 1/2] add auth mechanism --- README.md | 23 ++++++++++++ cmd/kroxy/main.go | 7 ++-- config/config.go | 22 +++++++++++- config/config_test.go | 65 +++++++++++++++++++++++++++++++++ dockerfiles/kroxy.yaml | 4 +++ proxy/conn.go | 50 ++++++++++++++++++-------- proxy/listener.go | 4 +++ upstream/conn.go | 81 ++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 238 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index e202279..c53ef31 100644 --- a/README.md +++ b/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. diff --git a/cmd/kroxy/main.go b/cmd/kroxy/main.go index 6195388..60d873f 100644 --- a/cmd/kroxy/main.go +++ b/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) diff --git a/config/config.go b/config/config.go index c230c17..5e2d340 100644 --- a/config/config.go +++ b/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 } diff --git a/config/config_test.go b/config/config_test.go index c48f9d4..0f81304 100644 --- a/config/config_test.go +++ b/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 { diff --git a/dockerfiles/kroxy.yaml b/dockerfiles/kroxy.yaml index 81d6b84..ca58965 100644 --- a/dockerfiles/kroxy.yaml +++ b/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 diff --git a/proxy/conn.go b/proxy/conn.go index 0bcbc07..99518d3 100644 --- a/proxy/conn.go +++ b/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,11 +201,29 @@ 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() @@ -213,7 +231,11 @@ func (c *conn) ensureUpstream() error { 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,7 +437,7 @@ 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() @@ -423,7 +445,7 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by 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,7 +468,7 @@ 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) } @@ -454,8 +476,8 @@ func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []by 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) } diff --git a/proxy/listener.go b/proxy/listener.go index 4e99fc9..dda3bca 100644 --- a/proxy/listener.go +++ b/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. diff --git a/upstream/conn.go b/upstream/conn.go index 482c36d..0de35ca 100644 --- a/upstream/conn.go +++ b/upstream/conn.go @@ -13,6 +13,8 @@ import ( "github.com/bubunyo/kroxy/protocol" "github.com/pkg/errors" "github.com/twmb/franz-go/pkg/kmsg" + "github.com/twmb/franz-go/pkg/sasl" + "github.com/twmb/franz-go/pkg/sasl/scram" ) // dialTimeout caps the time spent establishing the TCP connection and @@ -88,6 +90,85 @@ func DialForSCRAM(ctx context.Context, addr, mechanism string) (*Conn, error) { return c, nil } +// DialSCRAM opens a TCP connection to addr and authenticates with the given +// SCRAM mechanism, acting as the SCRAM client itself: it computes the client +// proof from username and the plaintext password. This is the PLAIN->SCRAM +// translation path — a client that authenticated to kroxy with SASL/PLAIN +// (so kroxy holds the plaintext password) is bridged to an upstream broker +// that only accepts SCRAM. It differs from DialForSCRAM, which relays an +// already-SCRAM client's frames verbatim without ever seeing the password. +// mechanism must be auth.MechanismSCRAMSHA256 or auth.MechanismSCRAMSHA512. +func DialSCRAM(ctx context.Context, addr, mechanism, username, password string) (*Conn, error) { + if !auth.IsSCRAMMechanism(mechanism) { + return nil, errors.Errorf("DialSCRAM: unsupported mechanism %q", mechanism) + } + c, err := dialAndNegotiate(ctx, addr, mechanism) + if err != nil { + return nil, errors.Wrap(err, "DialSCRAM") + } + if err := c.scramAuthenticate(ctx, mechanism, username, password); err != nil { + _ = c.nc.Close() + return nil, errors.Wrap(err, "DialSCRAM") + } + if err := c.nc.SetDeadline(time.Time{}); err != nil { + _ = c.nc.Close() + return nil, errors.Wrap(err, "DialSCRAM") + } + return c, nil +} + +// scramAuthenticate drives the SCRAM message exchange as the client. franz-go's +// scram mechanism computes the client-first / client-final messages and +// verifies the server's final signature; kroxy carries each message to the +// broker over SaslAuthenticate via RelaySASLAuthenticate and feeds the broker's +// reply back into the session until the exchange completes. +func (c *Conn) scramAuthenticate(ctx context.Context, mechanism, username, password string) error { + a := scram.Auth{User: username, Pass: password} + var mech sasl.Mechanism + switch mechanism { + case auth.MechanismSCRAMSHA256: + mech = a.AsSha256Mechanism() + case auth.MechanismSCRAMSHA512: + mech = a.AsSha512Mechanism() + default: + return errors.Errorf("scramAuthenticate: unsupported mechanism %q", mechanism) + } + + session, clientMsg, err := mech.Authenticate(ctx, c.nc.RemoteAddr().String()) + if err != nil { + return errors.Wrap(err, "scramAuthenticate") + } + + // SCRAM-SHA-256/512 are bounded to two round trips, but loop on the + // session's "done" signal rather than a fixed count so the franz-go + // mechanism stays the single source of truth for the exchange length. + for { + serverMsg, errCode, errMsg, rErr := c.RelaySASLAuthenticate(clientMsg) + if rErr != nil { + return errors.Wrap(rErr, "scramAuthenticate") + } + if errCode != 0 { + return errors.Errorf("scramAuthenticate: upstream error %d: %s", errCode, errMsg) + } + done, next, cErr := session.Challenge(serverMsg) + if cErr != nil { + return errors.Wrap(cErr, "scramAuthenticate") + } + if done { + if len(next) > 0 { + if _, errCode, errMsg, rErr = c.RelaySASLAuthenticate(next); rErr != nil { + return errors.Wrap(rErr, "scramAuthenticate") + } + if errCode != 0 { + return errors.Errorf("scramAuthenticate: upstream error %d: %s", errCode, errMsg) + } + } + return nil + } + clientMsg = next + } +} + // dialAndNegotiate opens the TCP connection, runs ApiVersions, and runs // SaslHandshake selecting the given mechanism. The returned Conn still has // the dial-deadline applied; the caller must clear it once initial From 5906b69d7452c60c595df0d217916e3550f51dba Mon Sep 17 00:00:00 2001 From: bubunyo nyavor Date: Thu, 11 Jun 2026 13:55:39 +0200 Subject: [PATCH 2/2] add auth mechanism --- proxy/translate_internal_test.go | 34 +++++ upstream/scram_test.go | 229 +++++++++++++++++++++++++++++++ 2 files changed, 263 insertions(+) create mode 100644 proxy/translate_internal_test.go create mode 100644 upstream/scram_test.go diff --git a/proxy/translate_internal_test.go b/proxy/translate_internal_test.go new file mode 100644 index 0000000..5999d88 --- /dev/null +++ b/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()) + }) + } +} diff --git a/upstream/scram_test.go b/upstream/scram_test.go new file mode 100644 index 0000000..31ca537 --- /dev/null +++ b/upstream/scram_test.go @@ -0,0 +1,229 @@ +package upstream + +import ( + "bytes" + "context" + "crypto/hmac" + "crypto/pbkdf2" + "crypto/sha256" + "encoding/base64" + "encoding/binary" + "fmt" + "net" + "strings" + "testing" + + "github.com/bubunyo/kroxy/auth" + "github.com/bubunyo/kroxy/protocol" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kmsg" +) + +// TestDialSCRAM drives the real franz-go SCRAM client (the same code kroxy uses +// to bridge a PLAIN client to a SCRAM broker) against a minimal in-process +// SCRAM-SHA-256 server. The server implements RFC 5802 verification, so a +// successful dial proves kroxy computed a correct client proof and verified the +// server's signature end to end. +func TestDialSCRAM(t *testing.T) { + t.Parallel() + + const ( + user = "tenantA" + pass = "tenantA-secret" + ) + + t.Run("success", func(t *testing.T) { + t.Parallel() + addr := startFakeSCRAMBroker(t, user, pass) + c, err := DialSCRAM(context.Background(), addr, auth.MechanismSCRAMSHA256, user, pass) + require.NoError(t, err) + require.NotNil(t, c) + _ = c.Close() + }) + + t.Run("wrong password is rejected by the broker", func(t *testing.T) { + t.Parallel() + addr := startFakeSCRAMBroker(t, user, pass) + c, err := DialSCRAM(context.Background(), addr, auth.MechanismSCRAMSHA256, user, "not-the-password") + require.Error(t, err) + assert.Nil(t, c) + }) + + t.Run("non-scram mechanism is rejected before dialing", func(t *testing.T) { + t.Parallel() + c, err := DialSCRAM(context.Background(), "127.0.0.1:0", auth.MechanismPlain, user, pass) + require.Error(t, err) + assert.Nil(t, c) + }) +} + +// startFakeSCRAMBroker listens on a loopback port and serves the upstream +// handshake (ApiVersions, SaslHandshake, SaslAuthenticate) for a single +// SCRAM-SHA-256 identity, giving each accepted connection fresh SCRAM state. +// It returns the listener address. +func startFakeSCRAMBroker(t *testing.T, user, pass string) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + nc, aerr := ln.Accept() + if aerr != nil { + return + } + go serveFakeBroker(t, nc, &scramServerSession{user: user, pass: pass, salt: []byte("kroxy-test-salt-0001"), iter: 4096}) + } + }() + return ln.Addr().String() +} + +func serveFakeBroker(t *testing.T, nc net.Conn, s *scramServerSession) { + defer func() { _ = nc.Close() }() + for { + frame, err := protocol.ReadFrame(nc) + if err != nil { + return + } + hdr, err := protocol.ParseRequestHeader(frame) + if err != nil { + t.Errorf("fake broker: parse header: %v", err) + return + } + body := frame[hdr.HeaderSize:] + switch hdr.APIKey { + case protocol.ApiVersionsKey: + resp := kmsg.NewPtrApiVersionsResponse() + resp.SetVersion(0) + writeBrokerResponse(t, nc, hdr.CorrelationID, resp.AppendTo(nil)) + case protocol.SaslHandshakeKey: + resp := kmsg.NewPtrSASLHandshakeResponse() + resp.SetVersion(1) + resp.SupportedMechanisms = []string{auth.MechanismSCRAMSHA256} + writeBrokerResponse(t, nc, hdr.CorrelationID, resp.AppendTo(nil)) + case protocol.SaslAuthenticateKey: + req := kmsg.NewPtrSASLAuthenticateRequest() + req.SetVersion(1) + if err := req.ReadFrom(body); err != nil { + t.Errorf("fake broker: read authenticate: %v", err) + return + } + out, code, msg := s.step(req.SASLAuthBytes) + resp := kmsg.NewPtrSASLAuthenticateResponse() + resp.SetVersion(1) + resp.SASLAuthBytes = out + resp.ErrorCode = code + if msg != "" { + resp.ErrorMessage = &msg + } + writeBrokerResponse(t, nc, hdr.CorrelationID, resp.AppendTo(nil)) + default: + return + } + } +} + +// writeBrokerResponse frames a v0/v1 (non-flexible) response: a 4-byte +// correlation id followed by the encoded message body, matching what the +// upstream Conn reads back during the handshake. +func writeBrokerResponse(t *testing.T, nc net.Conn, corrID int32, body []byte) { + out := make([]byte, 4, 4+len(body)) + binary.BigEndian.PutUint32(out, uint32(corrID)) + out = append(out, body...) + require.NoError(t, protocol.WriteFrame(nc, out)) +} + +// scramServerSession holds the per-connection SCRAM-SHA-256 server state and +// implements the two-message verification from RFC 5802 §5. +type scramServerSession struct { + user, pass string + salt []byte + iter int + clientFirst string + serverFirst string +} + +// step consumes one client SCRAM message and returns the server's reply, +// a Kafka error code (0 = ok, 58 = SASL auth failed), and an optional message. +func (s *scramServerSession) step(in []byte) ([]byte, int16, string) { + msg := string(in) + if s.serverFirst == "" { + bare := stripGS2Header(msg) + s.clientFirst = bare + cnonce := scramAttr(bare, "r") + if cnonce == "" { + return nil, 58, "missing client nonce" + } + s.serverFirst = fmt.Sprintf("r=%sservernonce,s=%s,i=%d", + cnonce, base64.StdEncoding.EncodeToString(s.salt), s.iter) + return []byte(s.serverFirst), 0, "" + } + + i := strings.Index(msg, ",p=") + if i < 0 { + return nil, 58, "missing proof" + } + withoutProof := msg[:i] + proof, err := base64.StdEncoding.DecodeString(msg[i+3:]) + if err != nil { + return nil, 58, "malformed proof" + } + authMessage := s.clientFirst + "," + s.serverFirst + "," + withoutProof + salted, err := pbkdf2.Key(sha256.New, s.pass, s.salt, s.iter, sha256.Size) + if err != nil { + return nil, 58, "key derivation failed" + } + clientKey := hmacSHA256(salted, []byte("Client Key")) + storedKey := sha256.Sum256(clientKey) + clientSig := hmacSHA256(storedKey[:], []byte(authMessage)) + recoveredKey := xorBytes(proof, clientSig) + recoveredStored := sha256.Sum256(recoveredKey) + if !bytes.Equal(recoveredStored[:], storedKey[:]) { + return nil, 58, "authentication failed" + } + serverKey := hmacSHA256(salted, []byte("Server Key")) + serverSig := hmacSHA256(serverKey, []byte(authMessage)) + return []byte("v=" + base64.StdEncoding.EncodeToString(serverSig)), 0, "" +} + +func hmacSHA256(key, data []byte) []byte { + m := hmac.New(sha256.New, key) + m.Write(data) + return m.Sum(nil) +} + +func xorBytes(a, b []byte) []byte { + out := make([]byte, len(a)) + for i := range a { + out[i] = a[i] ^ b[i] + } + return out +} + +// stripGS2Header returns the client-first-message-bare by dropping the +// gs2-cbind-flag and optional authzid (everything up to and including the +// second comma). +func stripGS2Header(s string) string { + i1 := strings.IndexByte(s, ',') + if i1 < 0 { + return s + } + rest := s[i1+1:] + i2 := strings.IndexByte(rest, ',') + if i2 < 0 { + return rest + } + return rest[i2+1:] +} + +// scramAttr returns the value of a "key=value" attribute in a comma-separated +// SCRAM message, or "" if absent. +func scramAttr(s, key string) string { + for _, part := range strings.Split(s, ",") { + if strings.HasPrefix(part, key+"=") { + return part[len(key)+1:] + } + } + return "" +}