diff --git a/README.md b/README.md
index e4c9a55..281e908 100644
--- a/README.md
+++ b/README.md
@@ -3,12 +3,12 @@
 A multi-tenant Kafka proxy written in Go.
 
 `kroxy` sits in front of a single Apache Kafka cluster and turns it into a
-multi-tenant service. It terminates SASL/PLAIN at the edge, uses the
-client's username to look up a tenant, and rewrites every topic, consumer
-group and transactional ID with a per-tenant prefix on the way to the
-upstream broker (and back). Each tenant sees a flat namespace that looks
-like its own dedicated cluster; the broker sees fully-qualified,
-prefixed names.
+multi-tenant service. It terminates SASL (PLAIN, SCRAM-SHA-256,
+SCRAM-SHA-512) at the edge, uses the client's username to look up a
+tenant, and rewrites every topic, consumer group and transactional ID
+with a per-tenant prefix on the way to the upstream broker (and back).
+Each tenant sees a flat namespace that looks like its own dedicated
+cluster; the broker sees fully-qualified, prefixed names.
 
 The proxy is a single static binary with no external dependencies beyond
 Kafka itself.
@@ -181,20 +181,33 @@ Notes:
 
 ## Authentication model
 
-kroxy is a SASL/PLAIN **pass-through**. The SASL username on the wire
-is the tenant ID; the password is forwarded verbatim to the tenant's
-upstream Kafka cluster, which is the sole auth authority. **kroxy
-stores no client secrets** and does not validate passwords itself.
+kroxy is a SASL **pass-through**. It supports three mechanisms:
+
+- **PLAIN** — single-shot. Username == tenant ID, password forwarded
+  verbatim to the upstream broker.
+- **SCRAM-SHA-256** — challenge/response. kroxy peeks only at the SASLname
+  in the SCRAM `client-first-message` (== tenant ID) for routing, then
+  relays every `SaslAuthenticate` frame between client and upstream
+  unchanged.
+- **SCRAM-SHA-512** — same model as SCRAM-SHA-256.
+
+The upstream Kafka cluster is the sole authentication authority for all
+three mechanisms. **kroxy stores no client secrets** and does not validate
+passwords or SCRAM proofs itself.
 
 Consequences:
 
 - Every tenant ID must be a real principal in the upstream broker
-  (declared in its JAAS file or auth backend, e.g. `kafka_jaas.conf`).
-- Unknown tenant IDs are rejected at the proxy before any upstream
-  dial, returning a SASL authentication failure.
-- Passwords are held in memory for the duration of the client
-  connection (to be able to reconnect to upstream on failure) and never
-  written to logs.
+  (declared in its JAAS file for PLAIN, or registered as SCRAM credentials
+  via `kafka-configs.sh --add-config 'SCRAM-SHA-256=[password=...]'` for
+  SCRAM).
+- Unknown tenant IDs are rejected at the proxy before any upstream dial,
+  returning a SASL authentication failure.
+- For PLAIN, the password is held in memory for the duration of the client
+  connection (to support upstream reconnection) and never written to logs.
+  For SCRAM, kroxy never observes the password at all.
+- SASL channel binding (`y`, `p=...`) is not supported — kroxy is not the
+  TLS terminator for the SCRAM exchange.
 
 The only thing kroxy needs to know about a tenant is the mapping
 `id → (topic_prefix, upstream)`.
@@ -278,7 +291,8 @@ deferred:
 
 - **No TLS** on either the client or upstream side. Run kroxy on a
   trusted network or behind a TLS-terminating sidecar.
-- **SASL/PLAIN only.** No SCRAM, no OAUTHBEARER, no mTLS, no Kerberos.
+- **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
   through but every tenant in the demo points at the same broker.
 - **No hot config reload.** Restart to pick up YAML changes; use the
diff --git a/auth/sasl_scram.go b/auth/sasl_scram.go
new file mode 100644
index 0000000..6ef1930
--- /dev/null
+++ b/auth/sasl_scram.go
@@ -0,0 +1,133 @@
+package auth
+
+import (
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+// SCRAM mechanisms advertised and accepted by the proxy alongside PLAIN.
+// kroxy implements these in pass-through "relay" mode: the SaslAuthenticate
+// payloads are forwarded verbatim between client and the upstream broker,
+// which is the sole authentication authority. kroxy peeks only at the first
+// client message in order to extract the SASLname (== tenant ID) for
+// routing.
+const (
+	MechanismSCRAMSHA256 = "SCRAM-SHA-256"
+	MechanismSCRAMSHA512 = "SCRAM-SHA-512"
+)
+
+// IsSCRAMMechanism reports whether mech is one of the SCRAM mechanisms
+// supported by the proxy.
+func IsSCRAMMechanism(mech string) bool {
+	return mech == MechanismSCRAMSHA256 || mech == MechanismSCRAMSHA512
+}
+
+// ParseSCRAMClientFirstUsername extracts the SASLname (== tenant ID) from a
+// SCRAM client-first-message as defined by RFC 5802 §7. The grammar we
+// accept is:
+//
+//	gs2-cbind-flag "," [ authzid ] "," "n=" saslname "," "r=" c-nonce ...
+//	gs2-cbind-flag = "n" | "y" | "p=..."
+//
+// kroxy does NOT support SASL channel binding, so only the "n" flag is
+// accepted; any "y" or "p=..." flag is rejected. authzid (if present) is
+// ignored. SASLname escapes "=2C" / "=3D" are decoded.
+func ParseSCRAMClientFirstUsername(payload []byte) (string, error) {
+	s := string(payload)
+
+	// gs2-cbind-flag.
+	cb, rest, ok := cutByte(s, ',')
+	if !ok {
+		return "", errors.New("ParseSCRAMClientFirstUsername: missing gs2 cbind-flag separator")
+	}
+	switch {
+	case cb == "n":
+		// no channel binding, ok.
+	case cb == "y" || strings.HasPrefix(cb, "p="):
+		return "", errors.New("ParseSCRAMClientFirstUsername: channel binding not supported")
+	default:
+		return "", errors.Errorf("ParseSCRAMClientFirstUsername: invalid gs2 cbind-flag %q", cb)
+	}
+
+	// optional authzid then "," then client-first-message-bare.
+	_, bare, ok := cutByte(rest, ',')
+	if !ok {
+		return "", errors.New("ParseSCRAMClientFirstUsername: missing authzid separator")
+	}
+
+	// client-first-message-bare = [reserved-mext ","] username "," nonce ["," extensions]
+	// Skip any leading m=... reserved-mext attribute.
+	if strings.HasPrefix(bare, "m=") {
+		_, after, ok := cutByte(bare, ',')
+		if !ok {
+			return "", errors.New("ParseSCRAMClientFirstUsername: malformed reserved-mext")
+		}
+		bare = after
+	}
+
+	if !strings.HasPrefix(bare, "n=") {
+		return "", errors.New("ParseSCRAMClientFirstUsername: missing n= attribute")
+	}
+	rest = bare[2:]
+	rawName, _, ok := cutByte(rest, ',')
+	if !ok {
+		return "", errors.New("ParseSCRAMClientFirstUsername: missing nonce separator")
+	}
+	if rawName == "" {
+		return "", errors.New("ParseSCRAMClientFirstUsername: empty username")
+	}
+	name, err := decodeSASLname(rawName)
+	if err != nil {
+		return "", errors.Wrap(err, "ParseSCRAMClientFirstUsername")
+	}
+	return name, nil
+}
+
+// cutByte splits s at the first occurrence of sep. It is a tiny helper to
+// avoid pulling in strings.Cut's allocation pattern repeatedly.
+func cutByte(s string, sep byte) (before, after string, found bool) {
+	if i := strings.IndexByte(s, sep); i >= 0 {
+		return s[:i], s[i+1:], true
+	}
+	return s, "", false
+}
+
+// decodeSASLname reverses the "=2C" / "=3D" escapes used by SCRAM SASLnames
+// (RFC 5802 §5.1). Any other "=XX" sequence, or a stray '=' or ',' in the
+// raw name, is rejected.
+func decodeSASLname(raw string) (string, error) {
+	if !strings.ContainsRune(raw, '=') {
+		// fast path: no escapes.
+		if strings.ContainsRune(raw, ',') {
+			return "", errors.New("decodeSASLname: unescaped comma")
+		}
+		return raw, nil
+	}
+	var b strings.Builder
+	b.Grow(len(raw))
+	for i := 0; i < len(raw); i++ {
+		c := raw[i]
+		switch c {
+		case ',':
+			return "", errors.New("decodeSASLname: unescaped comma")
+		case '=':
+			if i+2 >= len(raw) {
+				return "", errors.New("decodeSASLname: truncated escape")
+			}
+			esc := raw[i+1 : i+3]
+			switch esc {
+			case "2C":
+				b.WriteByte(',')
+			case "3D":
+				b.WriteByte('=')
+			default:
+				return "", errors.Errorf("decodeSASLname: invalid escape =%s", esc)
+			}
+			i += 2
+		default:
+			b.WriteByte(c)
+		}
+	}
+	return b.String(), nil
+}
diff --git a/auth/sasl_scram_test.go b/auth/sasl_scram_test.go
new file mode 100644
index 0000000..79fdda7
--- /dev/null
+++ b/auth/sasl_scram_test.go
@@ -0,0 +1,119 @@
+package auth_test
+
+import (
+	"testing"
+
+	"github.com/bubunyo/kroxy/auth"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestIsSCRAMMechanism(t *testing.T) {
+	t.Parallel()
+
+	assert.True(t, auth.IsSCRAMMechanism(auth.MechanismSCRAMSHA256))
+	assert.True(t, auth.IsSCRAMMechanism(auth.MechanismSCRAMSHA512))
+	assert.False(t, auth.IsSCRAMMechanism(auth.MechanismPlain))
+	assert.False(t, auth.IsSCRAMMechanism(""))
+	assert.False(t, auth.IsSCRAMMechanism("scram-sha-256"))
+}
+
+func TestParseSCRAMClientFirstUsername(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name    string
+		in      string
+		want    string
+		wantErr bool
+	}{
+		{
+			name: "no channel binding, no authzid",
+			in:   "n,,n=alice,r=fyko+d2lbbFgONRv9qkxdawL",
+			want: "alice",
+		},
+		{
+			name: "no channel binding, with authzid (ignored)",
+			in:   "n,a=admin,n=alice,r=abc",
+			want: "alice",
+		},
+		{
+			name: "escaped comma in name",
+			in:   "n,,n=al=2Cice,r=abc",
+			want: "al,ice",
+		},
+		{
+			name: "escaped equals in name",
+			in:   "n,,n=al=3Dice,r=abc",
+			want: "al=ice",
+		},
+		{
+			name: "with extensions after nonce",
+			in:   "n,,n=tenantA,r=abc,m=foo",
+			want: "tenantA",
+		},
+		{
+			name: "leading reserved-mext skipped",
+			in:   "n,,m=ignored,n=tenantA,r=abc",
+			want: "tenantA",
+		},
+		{
+			name:    "channel binding y rejected",
+			in:      "y,,n=alice,r=abc",
+			wantErr: true,
+		},
+		{
+			name:    "channel binding p= rejected",
+			in:      "p=tls-unique,,n=alice,r=abc",
+			wantErr: true,
+		},
+		{
+			name:    "missing gs2 cbind separator",
+			in:      "n",
+			wantErr: true,
+		},
+		{
+			name:    "missing authzid separator",
+			in:      "n,",
+			wantErr: true,
+		},
+		{
+			name:    "missing n= attribute",
+			in:      "n,,r=abc,n=alice",
+			wantErr: true,
+		},
+		{
+			name:    "missing nonce",
+			in:      "n,,n=alice",
+			wantErr: true,
+		},
+		{
+			name:    "empty username",
+			in:      "n,,n=,r=abc",
+			wantErr: true,
+		},
+		{
+			name:    "invalid escape",
+			in:      "n,,n=al=FFice,r=abc",
+			wantErr: true,
+		},
+		{
+			name:    "truncated escape",
+			in:      "n,,n=alice=2,r=abc",
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := auth.ParseSCRAMClientFirstUsername([]byte(tt.in))
+			if tt.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
diff --git a/dockerfiles/docker-compose.yml b/dockerfiles/docker-compose.yml
index 82b1e60..c3b49a2 100644
--- a/dockerfiles/docker-compose.yml
+++ b/dockerfiles/docker-compose.yml
@@ -25,8 +25,16 @@ services:
       KAFKA_INTER_BROKER_LISTENER_NAME: SASL_PLAINTEXT
       KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
       KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9094
-      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
+      KAFKA_SASL_ENABLED_MECHANISMS: PLAIN,SCRAM-SHA-256,SCRAM-SHA-512
       KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
+      # SCRAM is advertised by the broker but no SCRAM credentials are
+      # pre-provisioned in this demo stack. To bootstrap a SCRAM user, run:
+      #   docker exec kroxy-kafka /opt/kafka/bin/kafka-configs.sh \
+      #     --bootstrap-server kafka:9093 \
+      #     --command-config /etc/kafka/client.properties \
+      #     --alter --add-config 'SCRAM-SHA-256=[password=tenantApw]' \
+      #     --entity-type users --entity-name tenantA
+      # PLAIN works out of the box for tenantA / tenantApw and tenantB / tenantBpw.
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
       KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
diff --git a/dockerfiles/kafka_jaas.conf b/dockerfiles/kafka_jaas.conf
index 108cc93..df9ae11 100644
--- a/dockerfiles/kafka_jaas.conf
+++ b/dockerfiles/kafka_jaas.conf
@@ -5,6 +5,7 @@ KafkaServer {
     user_broker="brokerpw"
     user_tenantA="tenantApw"
     user_tenantB="tenantBpw";
+    org.apache.kafka.common.security.scram.ScramLoginModule required;
 };
 
 KafkaClient {
diff --git a/go.mod b/go.mod
index 09ee2b1..f37ec8e 100644
--- a/go.mod
+++ b/go.mod
@@ -36,6 +36,7 @@ require (
 	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/magiconair/properties v1.8.10 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
diff --git a/integration/kafka_test.go b/integration/kafka_test.go
index 0bc29a7..21e4135 100644
--- a/integration/kafka_test.go
+++ b/integration/kafka_test.go
@@ -17,7 +17,12 @@ import (
 // JAAS file declaring the SASL/PLAIN principals the test broker accepts.
 //
 // Inter-broker / admin uses "broker" / "brokerpw"; integration tests
-// authenticate as "tenantA", "tenantB", "carol".
+// authenticate as "tenantA", "tenantB", "carol" via PLAIN. SCRAM-SHA-256
+// and SCRAM-SHA-512 credentials for the same usernames are provisioned
+// once the broker is up by the starter script via kafka-configs.sh
+// (see copyStarterScript) using the broker admin principal over the
+// SASL_PLAINTEXT listener. SCRAM principals do not appear in this JAAS
+// file.
 const integrationJAAS = `KafkaServer {
     org.apache.kafka.common.security.plain.PlainLoginModule required
     username="broker"
@@ -26,6 +31,7 @@ const integrationJAAS = `KafkaServer {
     user_tenantA="tenantA"
     user_tenantB="tenantB"
     user_carol="carolpw";
+    org.apache.kafka.common.security.scram.ScramLoginModule required;
 };
 KafkaClient {
     org.apache.kafka.common.security.plain.PlainLoginModule required
@@ -56,12 +62,12 @@ func startKafkaSASL(ctx context.Context, t *testing.T) (string, func()) {
 			Env: map[string]string{
 				"KAFKA_NODE_ID":                                  "1",
 				"KAFKA_PROCESS_ROLES":                            "broker,controller",
-				"KAFKA_LISTENERS":                                "SASL_PLAINTEXT://0.0.0.0:9093,CONTROLLER://0.0.0.0:9094",
-				"KAFKA_LISTENER_SECURITY_PROTOCOL_MAP":           "SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT",
-				"KAFKA_INTER_BROKER_LISTENER_NAME":               "SASL_PLAINTEXT",
+				"KAFKA_LISTENERS":                                "SASL_PLAINTEXT://0.0.0.0:9093,INTERNAL://0.0.0.0:9092,CONTROLLER://0.0.0.0:9094",
+				"KAFKA_LISTENER_SECURITY_PROTOCOL_MAP":           "SASL_PLAINTEXT:SASL_PLAINTEXT,INTERNAL:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT",
+				"KAFKA_INTER_BROKER_LISTENER_NAME":               "INTERNAL",
 				"KAFKA_CONTROLLER_LISTENER_NAMES":                "CONTROLLER",
 				"KAFKA_CONTROLLER_QUORUM_VOTERS":                 "1@localhost:9094",
-				"KAFKA_SASL_ENABLED_MECHANISMS":                  "PLAIN",
+				"KAFKA_SASL_ENABLED_MECHANISMS":                  "PLAIN,SCRAM-SHA-256,SCRAM-SHA-512",
 				"KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL":     "PLAIN",
 				"KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR":         "1",
 				"KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR": "1",
@@ -82,7 +88,7 @@ func startKafkaSASL(ctx context.Context, t *testing.T) (string, func()) {
 			LifecycleHooks: []testcontainers.ContainerLifecycleHooks{{
 				PostStarts: []testcontainers.ContainerHook{copyStarterScript(externalPort)},
 			}},
-			WaitingFor: wait.ForLog("Kafka Server started").WithStartupTimeout(2 * time.Minute),
+			WaitingFor: wait.ForLog("KROXY_SCRAM_READY").WithStartupTimeout(3 * time.Minute),
 		},
 		Started: true,
 	}
@@ -118,8 +124,55 @@ func copyStarterScript(externalPort string) testcontainers.ContainerHook {
 		}
 		script := fmt.Sprintf(`#!/bin/bash
 set -e
-export KAFKA_ADVERTISED_LISTENERS="SASL_PLAINTEXT://%s:%s"
-exec /etc/kafka/docker/run
+export KAFKA_ADVERTISED_LISTENERS="SASL_PLAINTEXT://%s:%s,INTERNAL://localhost:9092"
+
+# Background the official wrapper so the broker comes up with PLAIN auth,
+# then provision SCRAM credentials over the running INTERNAL listener
+# using the broker admin principal. We can't bootstrap SCRAM users at
+# format time because the docker wrapper formats storage itself with its
+# own arguments and ignores any prior format. We use the INTERNAL
+# listener (advertised as localhost:9092 inside the container) so the
+# admin client can reach the broker without going through the host port
+# mapping.
+/etc/kafka/docker/run &
+BROKER_PID=$!
+
+# Admin client config: PLAIN as the inter-broker user.
+ADMIN_CFG=/tmp/admin.properties
+cat >"$ADMIN_CFG" <<EOF
+security.protocol=SASL_PLAINTEXT
+sasl.mechanism=PLAIN
+sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="broker" password="brokerpw";
+EOF
+
+# Wait for broker to accept admin requests on the internal listener.
+for i in $(seq 1 120); do
+  if /opt/kafka/bin/kafka-broker-api-versions.sh \
+      --bootstrap-server localhost:9092 \
+      --command-config "$ADMIN_CFG" >/dev/null 2>&1; then
+    break
+  fi
+  sleep 1
+done
+
+# Provision SCRAM credentials for each tenant on both digests. The
+# admin API rejects altering the same user twice in one request, so
+# each digest is added in a separate call.
+for entry in "tenantA:tenantA" "tenantB:tenantB" "carol:carolpw"; do
+  user="${entry%%:*}"
+  pw="${entry##*:}"
+  for mech in SCRAM-SHA-256 SCRAM-SHA-512; do
+    /opt/kafka/bin/kafka-configs.sh \
+        --bootstrap-server localhost:9092 \
+        --command-config "$ADMIN_CFG" \
+        --alter \
+        --add-config "$mech=[password=$pw]" \
+        --entity-type users --entity-name "$user"
+  done
+done
+
+echo "KROXY_SCRAM_READY"
+wait $BROKER_PID
 `, host, hostPort.Port())
 		return c.CopyToContainer(ctx, []byte(script), starterPath, 0o755)
 	}
diff --git a/integration/scram_test.go b/integration/scram_test.go
new file mode 100644
index 0000000..b5a9424
--- /dev/null
+++ b/integration/scram_test.go
@@ -0,0 +1,136 @@
+//go:build integration
+
+package integration_test
+
+import (
+	"context"
+	"crypto/sha256"
+	"crypto/sha512"
+	"fmt"
+	"hash"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/twmb/franz-go/pkg/kmsg"
+	"github.com/twmb/franz-go/pkg/sasl/plain"
+	"github.com/twmb/franz-go/pkg/sasl/scram"
+)
+
+// newSCRAMClient builds a kgo.Client that authenticates via SCRAM. The
+// mechanism is selected from the digest size returned by h: SHA-256 or
+// SHA-512.
+func newSCRAMClient(t *testing.T, addr, user, pw string, h func() hash.Hash, extra ...kgo.Opt) *kgo.Client {
+	t.Helper()
+	a := scram.Auth{User: user, Pass: pw}
+	var saslOpt kgo.Opt
+	switch h().Size() {
+	case sha256.Size:
+		saslOpt = kgo.SASL(a.AsSha256Mechanism())
+	case sha512.Size:
+		saslOpt = kgo.SASL(a.AsSha512Mechanism())
+	default:
+		t.Fatalf("unsupported hash size %d", h().Size())
+	}
+
+	opts := []kgo.Opt{
+		kgo.SeedBrokers(addr),
+		saslOpt,
+		kgo.RequestTimeoutOverhead(15 * time.Second),
+		kgo.MetadataMinAge(100 * time.Millisecond),
+	}
+	opts = append(opts, extra...)
+	cl, err := kgo.NewClient(opts...)
+	require.NoError(t, err)
+	return cl
+}
+
+// TestEndToEnd_SCRAMSHA256 verifies a full produce/consume round-trip
+// through kroxy using SCRAM-SHA-256 against the upstream broker, which is
+// the sole authentication authority.
+func TestEndToEnd_SCRAMSHA256(t *testing.T) {
+	runSCRAMEndToEnd(t, sha256.New, "scram256-events")
+}
+
+// TestEndToEnd_SCRAMSHA512 mirrors the SHA-256 test for SHA-512.
+func TestEndToEnd_SCRAMSHA512(t *testing.T) {
+	runSCRAMEndToEnd(t, sha512.New, "scram512-events")
+}
+
+func runSCRAMEndToEnd(t *testing.T, h func() hash.Hash, topic string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute)
+	defer cancel()
+
+	upstream, stopK := startKafkaSASL(ctx, t)
+	t.Cleanup(stopK)
+
+	proxyAddr, stop := startProxy(t, upstream)
+	defer stop()
+
+	prod := newSCRAMClient(t, proxyAddr, tenantA, tenantAPw, h,
+		kgo.AllowAutoTopicCreation(),
+		kgo.DefaultProduceTopic(topic),
+	)
+	defer prod.Close()
+
+	const recordCount = 5
+	pCtx, pCancel := context.WithTimeout(ctx, 60*time.Second)
+	defer pCancel()
+	for i := 0; i < recordCount; i++ {
+		r := &kgo.Record{Value: []byte(fmt.Sprintf("scram-%d", i))}
+		results := prod.ProduceSync(pCtx, r)
+		require.NoError(t, results.FirstErr(), "produce failed at %d", i)
+	}
+
+	cons := newSCRAMClient(t, proxyAddr, tenantA, tenantAPw, h,
+		kgo.ConsumerGroup(topic+"-readers"),
+		kgo.ConsumeTopics(topic),
+		kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()),
+	)
+	defer cons.Close()
+
+	got := make([]string, 0, recordCount)
+	cCtx, cCancel := context.WithTimeout(ctx, 60*time.Second)
+	defer cCancel()
+	for len(got) < recordCount {
+		fetches := cons.PollFetches(cCtx)
+		require.False(t, fetches.IsClientClosed())
+		fetches.EachError(func(_ string, _ int32, err error) {
+			t.Fatalf("consumer error: %v", err)
+		})
+		fetches.EachRecord(func(r *kgo.Record) {
+			got = append(got, string(r.Value))
+			assert.Equal(t, topic, r.Topic)
+		})
+	}
+
+	want := make([]string, recordCount)
+	for i := range want {
+		want[i] = fmt.Sprintf("scram-%d", i)
+	}
+	assert.ElementsMatch(t, want, got)
+
+	// Verify the broker stored the prefixed topic name.
+	direct, err := kgo.NewClient(
+		kgo.SeedBrokers(upstream),
+		kgo.SASL(plain.Auth{User: "broker", Pass: "brokerpw"}.AsMechanism()),
+		kgo.MetadataMinAge(100*time.Millisecond),
+	)
+	require.NoError(t, err)
+	defer direct.Close()
+
+	mdReq := kmsg.NewPtrMetadataRequest()
+	mdResp, err := mdReq.RequestWith(ctx, direct)
+	require.NoError(t, err)
+
+	var found bool
+	for _, top := range mdResp.Topics {
+		if top.Topic != nil && *top.Topic == tenantA+"."+topic {
+			found = true
+			break
+		}
+	}
+	assert.True(t, found, "expected %s.%s on the broker", tenantA, topic)
+}
diff --git a/observability/metrics.go b/observability/metrics.go
index 4ca07e8..8a180dd 100644
--- a/observability/metrics.go
+++ b/observability/metrics.go
@@ -22,6 +22,7 @@ type Metrics struct {
 	RequestDuration    *prometheus.HistogramVec
 	UpstreamErrorTotal *prometheus.CounterVec
 	ResolverCallsTotal *prometheus.CounterVec
+	SaslHandshakeTotal *prometheus.CounterVec
 }
 
 // NewMetrics builds and registers the proxy's Prometheus metrics on a
@@ -55,6 +56,11 @@ func NewMetrics() *Metrics {
 			Namespace: "kroxy", Name: "resolver_calls_total",
 			Help: "Total number of resolver lookups, labelled by result.",
 		}, []string{"result"}),
+		SaslHandshakeTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Namespace: "kroxy", Name: "sasl_handshakes_total",
+			Help: "Total SASL handshake outcomes labelled by mechanism and result " +
+				"(ok, unsupported, malformed, unauthorized, upstream_error).",
+		}, []string{"mechanism", "result"}),
 	}
 	reg.MustRegister(
 		m.ConnectionsActive,
@@ -63,6 +69,7 @@ func NewMetrics() *Metrics {
 		m.RequestDuration,
 		m.UpstreamErrorTotal,
 		m.ResolverCallsTotal,
+		m.SaslHandshakeTotal,
 	)
 	return m
 }
diff --git a/observability/metrics_test.go b/observability/metrics_test.go
index 328449d..217e3e7 100644
--- a/observability/metrics_test.go
+++ b/observability/metrics_test.go
@@ -22,6 +22,8 @@ func TestMetrics_Handler_ExposesProxyMetrics(t *testing.T) {
 	m.ObserveRequest(3, "tenantA", 5*time.Millisecond)
 	m.UpstreamErrorTotal.WithLabelValues("dial").Inc()
 	m.ResolverCallsTotal.WithLabelValues("ok").Inc()
+	m.SaslHandshakeTotal.WithLabelValues("PLAIN", "ok").Inc()
+	m.SaslHandshakeTotal.WithLabelValues("SCRAM-SHA-256", "unauthorized").Inc()
 
 	srv := httptest.NewServer(m.Handler())
 	defer srv.Close()
@@ -41,6 +43,8 @@ func TestMetrics_Handler_ExposesProxyMetrics(t *testing.T) {
 		`kroxy_requests_total{api_key="3",tenant="tenantA"} 1`,
 		`kroxy_upstream_errors_total{kind="dial"} 1`,
 		`kroxy_resolver_calls_total{result="ok"} 1`,
+		`kroxy_sasl_handshakes_total{mechanism="PLAIN",result="ok"} 1`,
+		`kroxy_sasl_handshakes_total{mechanism="SCRAM-SHA-256",result="unauthorized"} 1`,
 	} {
 		assert.True(t, strings.Contains(out, want), "metrics output missing %q\n---\n%s", want, out)
 	}
diff --git a/proxy/conn.go b/proxy/conn.go
index 905bd6a..0bcbc07 100644
--- a/proxy/conn.go
+++ b/proxy/conn.go
@@ -32,6 +32,7 @@ type connState int
 const (
 	stateAwaitHandshake connState = iota
 	stateAwaitAuth
+	stateRelaySaslInFlight
 	stateAuthenticated
 )
 
@@ -43,10 +44,12 @@ type conn struct {
 	metrics  *observability.Metrics
 	log      *slog.Logger
 
-	state    connState
-	tenant   resolver.Tenant
-	password string
-	upstream *upstream.Conn
+	state                connState
+	mechanism            string
+	tenant               resolver.Tenant
+	password             string // populated only on the PLAIN path
+	upstream             *upstream.Conn
+	scramRoundsCompleted int
 }
 
 func newConn(ctx context.Context, nc net.Conn, r resolver.Resolver, cfg ServerConfig, m *observability.Metrics, log *slog.Logger) *conn {
@@ -260,23 +263,58 @@ func (c *conn) handleSaslHandshake(hdr protocol.RequestHeader, body []byte) erro
 	}
 	resp := kmsg.NewPtrSASLHandshakeResponse()
 	resp.SetVersion(hdr.APIVersion)
-	resp.SupportedMechanisms = []string{auth.MechanismPlain}
+	resp.SupportedMechanisms = []string{
+		auth.MechanismPlain,
+		auth.MechanismSCRAMSHA256,
+		auth.MechanismSCRAMSHA512,
+	}
 	switch {
 	case c.state != stateAwaitHandshake:
 		resp.ErrorCode = errIllegalSaslState
-	case req.Mechanism != auth.MechanismPlain:
+		c.observeHandshake(req.Mechanism, "illegal_state")
+	case req.Mechanism != auth.MechanismPlain && !auth.IsSCRAMMechanism(req.Mechanism):
 		resp.ErrorCode = errUnsupportedSaslMech
+		c.observeHandshake(req.Mechanism, "unsupported")
 	default:
+		c.mechanism = req.Mechanism
 		c.state = stateAwaitAuth
 	}
 	return c.writeResponse(hdr, resp)
 }
 
+// observeHandshake records the result of a SASL handshake or authenticate
+// step against the SaslHandshakeTotal counter, if metrics are enabled.
+//
+// The mechanism label is bounded to a fixed allow-list (PLAIN,
+// SCRAM-SHA-256, SCRAM-SHA-512); any other value — including the empty
+// string and arbitrary client-supplied strings from a SaslHandshake
+// request — is normalized to "unknown" so a hostile or buggy client
+// cannot create unbounded Prometheus label cardinality.
+func (c *conn) observeHandshake(mech, result string) {
+	if c.metrics == nil {
+		return
+	}
+	switch mech {
+	case auth.MechanismPlain, auth.MechanismSCRAMSHA256, auth.MechanismSCRAMSHA512:
+		// allowed
+	default:
+		mech = "unknown"
+	}
+	c.metrics.SaslHandshakeTotal.WithLabelValues(mech, result).Inc()
+}
+
 func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) error {
+	if auth.IsSCRAMMechanism(c.mechanism) {
+		return c.handleSaslAuthenticateSCRAM(hdr, body)
+	}
+	return c.handleSaslAuthenticatePlain(hdr, body)
+}
+
+func (c *conn) handleSaslAuthenticatePlain(hdr protocol.RequestHeader, body []byte) error {
 	req := kmsg.NewPtrSASLAuthenticateRequest()
 	req.SetVersion(hdr.APIVersion)
 	if err := req.ReadFrom(body); err != nil {
-		return errors.Wrap(err, "handleSaslAuthenticate")
+		return errors.Wrap(err, "handleSaslAuthenticatePlain")
 	}
 	resp := kmsg.NewPtrSASLAuthenticateResponse()
 	resp.SetVersion(hdr.APIVersion)
@@ -285,6 +323,7 @@ func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) e
 		resp.ErrorCode = errIllegalSaslState
 		msg := "SASL handshake required"
 		resp.ErrorMessage = &msg
+		c.observeHandshake(auth.MechanismPlain, "illegal_state")
 		return c.writeResponse(hdr, resp)
 	}
 
@@ -293,6 +332,7 @@ func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) e
 		resp.ErrorCode = errSaslAuthFailed
 		msg := "malformed PLAIN payload"
 		resp.ErrorMessage = &msg
+		c.observeHandshake(auth.MechanismPlain, "malformed")
 		return c.writeResponse(hdr, resp)
 	}
 
@@ -308,6 +348,7 @@ func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) e
 		resp.ErrorCode = errSaslAuthFailed
 		msg := "authentication failed"
 		resp.ErrorMessage = &msg
+		c.observeHandshake(auth.MechanismPlain, "unauthorized")
 		c.log.InfoContext(c.ctx, "sasl auth failed", "tenant_id", creds.Username, "err", err)
 		return c.writeResponse(hdr, resp)
 	}
@@ -318,7 +359,104 @@ func (c *conn) handleSaslAuthenticate(hdr protocol.RequestHeader, body []byte) e
 	c.tenant = tenant
 	c.password = creds.Password
 	c.state = stateAuthenticated
-	c.log.InfoContext(c.ctx, "sasl auth ok", "tenant_id", tenant.ID)
+	c.observeHandshake(auth.MechanismPlain, "ok")
+	c.log.InfoContext(c.ctx, "sasl auth ok", "tenant_id", tenant.ID, "mechanism", auth.MechanismPlain)
+	return c.writeResponse(hdr, resp)
+}
+
+// handleSaslAuthenticateSCRAM relays SCRAM SaslAuthenticate frames between
+// the client and the upstream broker. On the first message kroxy parses the
+// SASLname (== tenant ID) from the SCRAM client-first-message in order to
+// resolve the tenant and dial the correct upstream; subsequent messages are
+// forwarded verbatim. kroxy never inspects nonces, salts, or proofs — the
+// upstream broker is the sole authentication authority.
+func (c *conn) handleSaslAuthenticateSCRAM(hdr protocol.RequestHeader, body []byte) error {
+	req := kmsg.NewPtrSASLAuthenticateRequest()
+	req.SetVersion(hdr.APIVersion)
+	if err := req.ReadFrom(body); err != nil {
+		return errors.Wrap(err, "handleSaslAuthenticateSCRAM")
+	}
+	resp := kmsg.NewPtrSASLAuthenticateResponse()
+	resp.SetVersion(hdr.APIVersion)
+
+	if c.state != stateAwaitAuth && c.state != stateRelaySaslInFlight {
+		resp.ErrorCode = errIllegalSaslState
+		msg := "SASL handshake required"
+		resp.ErrorMessage = &msg
+		c.observeHandshake(c.mechanism, "illegal_state")
+		return c.writeResponse(hdr, resp)
+	}
+
+	// First SCRAM message: extract username, resolve tenant, dial upstream.
+	if c.state == stateAwaitAuth {
+		username, err := auth.ParseSCRAMClientFirstUsername(req.SASLAuthBytes)
+		if err != nil {
+			resp.ErrorCode = errSaslAuthFailed
+			msg := "malformed SCRAM client-first-message"
+			resp.ErrorMessage = &msg
+			c.observeHandshake(c.mechanism, "malformed")
+			c.log.InfoContext(c.ctx, "sasl scram parse failed", "err", err)
+			return c.writeResponse(hdr, resp)
+		}
+		tenant, err := c.resolver.Get(c.ctx, username)
+		if err != nil {
+			if c.metrics != nil {
+				c.metrics.ResolverCallsTotal.WithLabelValues("unauthorized").Inc()
+			}
+			resp.ErrorCode = errSaslAuthFailed
+			msg := "authentication failed"
+			resp.ErrorMessage = &msg
+			c.observeHandshake(c.mechanism, "unauthorized")
+			c.log.InfoContext(c.ctx, "sasl auth failed", "tenant_id", username, "err", err)
+			return c.writeResponse(hdr, resp)
+		}
+		if c.metrics != nil {
+			c.metrics.ResolverCallsTotal.WithLabelValues("ok").Inc()
+		}
+		c.tenant = tenant
+
+		up, dErr := upstream.DialForSCRAM(c.ctx, tenant.Upstream, c.mechanism)
+		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.log.WarnContext(c.ctx, "scram upstream dial failed", "tenant_id", tenant.ID, "err", dErr)
+			return c.writeResponse(hdr, resp)
+		}
+		c.upstream = up
+		c.state = stateRelaySaslInFlight
+	}
+
+	respBytes, errCode, errMsg, rErr := c.upstream.RelaySASLAuthenticate(req.SASLAuthBytes)
+	if rErr != nil {
+		if c.metrics != nil {
+			c.metrics.UpstreamErrorTotal.WithLabelValues("scram_relay").Inc()
+		}
+		c.observeHandshake(c.mechanism, "upstream_error")
+		return errors.Wrap(rErr, "handleSaslAuthenticateSCRAM")
+	}
+	resp.SASLAuthBytes = respBytes
+	resp.ErrorCode = errCode
+	if errMsg != "" {
+		m := errMsg
+		resp.ErrorMessage = &m
+	}
+	if errCode != 0 {
+		c.observeHandshake(c.mechanism, "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)
+	}
 	return c.writeResponse(hdr, resp)
 }
 
diff --git a/proxy/conn_internal_test.go b/proxy/conn_internal_test.go
new file mode 100644
index 0000000..1237165
--- /dev/null
+++ b/proxy/conn_internal_test.go
@@ -0,0 +1,44 @@
+package proxy
+
+import (
+	"testing"
+
+	"github.com/bubunyo/kroxy/observability"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/assert"
+)
+
+// TestObserveHandshake_BoundedMechanismLabel verifies that the mechanism
+// label on SaslHandshakeTotal is restricted to the supported allow-list.
+// Arbitrary client-supplied strings (e.g., from a malicious or buggy
+// client probing SaslHandshake mechanisms) must collapse to "unknown" so
+// they cannot create unbounded Prometheus label cardinality. Regression
+// test for a Copilot review finding on the SCRAM relay PR.
+func TestObserveHandshake_BoundedMechanismLabel(t *testing.T) {
+	t.Parallel()
+
+	m := observability.NewMetrics()
+	c := &conn{metrics: m}
+
+	c.observeHandshake("PLAIN", "ok")
+	c.observeHandshake("SCRAM-SHA-256", "ok")
+	c.observeHandshake("SCRAM-SHA-512", "ok")
+	c.observeHandshake("FOOBAR-1234", "unsupported")
+	c.observeHandshake("FOOBAR-5678", "unsupported")
+	c.observeHandshake("", "illegal_state")
+
+	cv := m.SaslHandshakeTotal
+
+	assert.Equal(t, float64(1), testutil.ToFloat64(cv.WithLabelValues("PLAIN", "ok")))
+	assert.Equal(t, float64(1), testutil.ToFloat64(cv.WithLabelValues("SCRAM-SHA-256", "ok")))
+	assert.Equal(t, float64(1), testutil.ToFloat64(cv.WithLabelValues("SCRAM-SHA-512", "ok")))
+
+	// Two separate hostile mechanism strings + one empty mechanism all
+	// collapse onto the single "unknown" series (split by result label).
+	assert.Equal(t, float64(2), testutil.ToFloat64(cv.WithLabelValues("unknown", "unsupported")))
+	assert.Equal(t, float64(1), testutil.ToFloat64(cv.WithLabelValues("unknown", "illegal_state")))
+
+	// And the raw client-supplied strings must NOT appear as label values.
+	assert.Equal(t, float64(0), testutil.ToFloat64(cv.WithLabelValues("FOOBAR-1234", "unsupported")))
+	assert.Equal(t, float64(0), testutil.ToFloat64(cv.WithLabelValues("FOOBAR-5678", "unsupported")))
+}
diff --git a/proxy/conn_scram_test.go b/proxy/conn_scram_test.go
new file mode 100644
index 0000000..049373a
--- /dev/null
+++ b/proxy/conn_scram_test.go
@@ -0,0 +1,182 @@
+package proxy_test
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/bubunyo/kroxy/protocol"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/twmb/franz-go/pkg/kmsg"
+)
+
+func scramHandshake(t *testing.T, c net.Conn, mechanism string, cid *int32) {
+	t.Helper()
+	hsReq := kmsg.NewPtrSASLHandshakeRequest()
+	hsReq.SetVersion(1)
+	hsReq.Mechanism = mechanism
+	*cid++
+	sendRequest(t, c, hsReq, *cid, "test")
+	hsResp := kmsg.NewPtrSASLHandshakeResponse()
+	hsResp.SetVersion(1)
+	_ = recvResponse(t, c, hsResp, protocol.SaslHandshakeKey, 1)
+	require.Equal(t, int16(0), hsResp.ErrorCode, "handshake rejected: mechs=%v", hsResp.SupportedMechanisms)
+}
+
+func sendSaslAuth(t *testing.T, c net.Conn, payload []byte, cid *int32) *kmsg.SASLAuthenticateResponse {
+	t.Helper()
+	authReq := kmsg.NewPtrSASLAuthenticateRequest()
+	authReq.SetVersion(1)
+	authReq.SASLAuthBytes = payload
+	*cid++
+	sendRequest(t, c, authReq, *cid, "test")
+	authResp := kmsg.NewPtrSASLAuthenticateResponse()
+	authResp.SetVersion(1)
+	_ = recvResponse(t, c, authResp, protocol.SaslAuthenticateKey, 1)
+	return authResp
+}
+
+func TestSCRAM_HandshakeAdvertisesAllMechanisms(t *testing.T) {
+	t.Parallel()
+
+	addr, stop := startTestServer(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)))
+
+	hsReq := kmsg.NewPtrSASLHandshakeRequest()
+	hsReq.SetVersion(1)
+	hsReq.Mechanism = "SCRAM-SHA-256"
+	sendRequest(t, c, hsReq, 1, "test")
+	hsResp := kmsg.NewPtrSASLHandshakeResponse()
+	hsResp.SetVersion(1)
+	_ = recvResponse(t, c, hsResp, protocol.SaslHandshakeKey, 1)
+	assert.Equal(t, int16(0), hsResp.ErrorCode)
+	assert.Contains(t, hsResp.SupportedMechanisms, "PLAIN")
+	assert.Contains(t, hsResp.SupportedMechanisms, "SCRAM-SHA-256")
+	assert.Contains(t, hsResp.SupportedMechanisms, "SCRAM-SHA-512")
+}
+
+func TestSCRAM_RelayHappyPath(t *testing.T) {
+	t.Parallel()
+
+	upstreamReplies := [][]byte{
+		[]byte("server-first-message-stub"),
+		[]byte("v=server-final-signature-stub"),
+	}
+	broker := newFakeBroker(t)
+	broker.scramResponses = upstreamReplies
+	defer broker.close()
+
+	addr, stop := startTestServerWithUpstream(t, broker.addr)
+	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)))
+
+	var cid int32
+	scramHandshake(t, c, "SCRAM-SHA-256", &cid)
+
+	first := []byte("n,,n=alice,r=clientnonce")
+	r1 := sendSaslAuth(t, c, first, &cid)
+	assert.Equal(t, int16(0), r1.ErrorCode)
+	assert.Equal(t, upstreamReplies[0], r1.SASLAuthBytes)
+
+	final := []byte("c=biws,r=clientnoncesrvnonce,p=proof")
+	r2 := sendSaslAuth(t, c, final, &cid)
+	assert.Equal(t, int16(0), r2.ErrorCode)
+	assert.Equal(t, upstreamReplies[1], r2.SASLAuthBytes)
+
+	broker.mu.Lock()
+	got := append([][]byte(nil), broker.receivedSaslBytes...)
+	broker.mu.Unlock()
+	require.Len(t, got, 2)
+	assert.Equal(t, first, got[0])
+	assert.Equal(t, final, got[1])
+}
+
+func TestSCRAM_UnknownTenantRejected(t *testing.T) {
+	t.Parallel()
+
+	broker := newFakeBroker(t)
+	broker.scramResponses = [][]byte{[]byte("server-first")}
+	defer broker.close()
+
+	addr, stop := startTestServerWithUpstream(t, broker.addr)
+	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)))
+
+	var cid int32
+	scramHandshake(t, c, "SCRAM-SHA-512", &cid)
+
+	r := sendSaslAuth(t, c, []byte("n,,n=ghost,r=nonce"), &cid)
+	assert.NotEqual(t, int16(0), r.ErrorCode)
+
+	// Upstream must not have received any SaslAuthenticate.
+	broker.mu.Lock()
+	defer broker.mu.Unlock()
+	assert.Empty(t, broker.receivedSaslBytes)
+}
+
+func TestSCRAM_ChannelBindingRejected(t *testing.T) {
+	t.Parallel()
+
+	broker := newFakeBroker(t)
+	broker.scramResponses = [][]byte{[]byte("ignored")}
+	defer broker.close()
+
+	addr, stop := startTestServerWithUpstream(t, broker.addr)
+	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)))
+
+	var cid int32
+	scramHandshake(t, c, "SCRAM-SHA-256", &cid)
+
+	r := sendSaslAuth(t, c, []byte("y,,n=alice,r=nonce"), &cid)
+	assert.NotEqual(t, int16(0), r.ErrorCode)
+
+	broker.mu.Lock()
+	defer broker.mu.Unlock()
+	assert.Empty(t, broker.receivedSaslBytes)
+}
+
+func TestSCRAM_UpstreamRejectsFinal(t *testing.T) {
+	t.Parallel()
+
+	broker := newFakeBroker(t)
+	broker.scramResponses = [][]byte{[]byte("server-first"), []byte("ignored")}
+	broker.scramFailOnRound = 2
+	broker.scramFailCode = 58
+	defer broker.close()
+
+	addr, stop := startTestServerWithUpstream(t, broker.addr)
+	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)))
+
+	var cid int32
+	scramHandshake(t, c, "SCRAM-SHA-256", &cid)
+
+	r1 := sendSaslAuth(t, c, []byte("n,,n=alice,r=nonce"), &cid)
+	require.Equal(t, int16(0), r1.ErrorCode)
+
+	r2 := sendSaslAuth(t, c, []byte("c=biws,r=nonce,p=proof"), &cid)
+	assert.Equal(t, int16(58), r2.ErrorCode)
+}
diff --git a/proxy/upstream_test.go b/proxy/upstream_test.go
index 91e7ceb..996eef7 100644
--- a/proxy/upstream_test.go
+++ b/proxy/upstream_test.go
@@ -44,6 +44,15 @@ type fakeBroker struct {
 	lastCoordKey string
 	// Last JoinGroup group ID the broker saw, post-decode.
 	lastJoinGroup string
+
+	// SCRAM relay support. When scramResponses is non-nil, the broker
+	// returns these payloads on successive SaslAuthenticate requests
+	// instead of the single-shot PLAIN behaviour. scramFailOnRound is
+	// 1-based; 0 disables the failure injection.
+	scramResponses    [][]byte
+	scramFailOnRound  int
+	scramFailCode     int16
+	receivedSaslBytes [][]byte
 }
 
 func newFakeBroker(t *testing.T) *fakeBroker {
@@ -73,6 +82,7 @@ func (b *fakeBroker) serve() {
 func (b *fakeBroker) handle(c net.Conn) {
 	defer c.Close()
 	authed := false
+	scramRound := 0
 	for {
 		frame, err := protocol.ReadFrame(c)
 		if err != nil {
@@ -96,7 +106,7 @@ func (b *fakeBroker) handle(c net.Conn) {
 		case protocol.SaslHandshakeKey:
 			resp := kmsg.NewPtrSASLHandshakeResponse()
 			resp.SetVersion(hdr.APIVersion)
-			resp.SupportedMechanisms = []string{"PLAIN"}
+			resp.SupportedMechanisms = []string{"PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"}
 			b.write(c, resp, hdr)
 		case protocol.SaslAuthenticateKey:
 			req := kmsg.NewPtrSASLAuthenticateRequest()
@@ -104,11 +114,28 @@ func (b *fakeBroker) handle(c net.Conn) {
 			_ = req.ReadFrom(body)
 			b.mu.Lock()
 			b.gotCreds = string(req.SASLAuthBytes)
+			b.receivedSaslBytes = append(b.receivedSaslBytes, append([]byte(nil), req.SASLAuthBytes...))
+			scramMode := b.scramResponses != nil
 			b.mu.Unlock()
+
 			resp := kmsg.NewPtrSASLAuthenticateResponse()
 			resp.SetVersion(hdr.APIVersion)
+			if scramMode {
+				scramRound++
+				if b.scramFailOnRound > 0 && scramRound == b.scramFailOnRound {
+					resp.ErrorCode = b.scramFailCode
+					m := "fake broker rejected"
+					resp.ErrorMessage = &m
+				} else if scramRound-1 < len(b.scramResponses) {
+					resp.SASLAuthBytes = b.scramResponses[scramRound-1]
+				}
+				if scramRound >= len(b.scramResponses) {
+					authed = true
+				}
+			} else {
+				authed = true
+			}
 			b.write(c, resp, hdr)
-			authed = true
 		default:
 			if !authed {
 				return
diff --git a/upstream/conn.go b/upstream/conn.go
index 96cbcbe..482c36d 100644
--- a/upstream/conn.go
+++ b/upstream/conn.go
@@ -49,24 +49,68 @@ func (c *Conn) applyRequestDeadline() error {
 // verbatim — kroxy does not validate them; the upstream broker is the auth
 // authority.
 func Dial(ctx context.Context, addr, username, password string) (*Conn, error) {
+	c, err := dialAndNegotiate(ctx, addr, auth.MechanismPlain)
+	if err != nil {
+		return nil, errors.Wrap(err, "Dial")
+	}
+	if err := c.plainAuthenticate(username, password); err != nil {
+		_ = c.nc.Close()
+		return nil, errors.Wrap(err, "Dial")
+	}
+	if err := c.nc.SetDeadline(time.Time{}); err != nil {
+		_ = c.nc.Close()
+		return nil, errors.Wrap(err, "Dial")
+	}
+	return c, nil
+}
+
+// DialForSCRAM opens a TCP connection to addr and performs ApiVersions +
+// SaslHandshake selecting mechanism, but does NOT complete the
+// SaslAuthenticate exchange. Callers drive the SCRAM message rounds via
+// RelaySASLAuthenticate, which forwards the client-supplied payloads
+// verbatim. mechanism must be one of auth.MechanismSCRAMSHA256 /
+// auth.MechanismSCRAMSHA512.
+func DialForSCRAM(ctx context.Context, addr, mechanism string) (*Conn, error) {
+	if !auth.IsSCRAMMechanism(mechanism) {
+		return nil, errors.Errorf("DialForSCRAM: unsupported mechanism %q", mechanism)
+	}
+	c, err := dialAndNegotiate(ctx, addr, mechanism)
+	if err != nil {
+		return nil, errors.Wrap(err, "DialForSCRAM")
+	}
+	// Leave the deadline cleared; RelaySASLAuthenticate applies a
+	// per-request deadline via applyRequestDeadline before each round
+	// trip so a stalled upstream cannot block indefinitely.
+	if err := c.nc.SetDeadline(time.Time{}); err != nil {
+		_ = c.nc.Close()
+		return nil, errors.Wrap(err, "DialForSCRAM")
+	}
+	return c, nil
+}
+
+// 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
+// authentication is complete.
+func dialAndNegotiate(ctx context.Context, addr, mechanism string) (*Conn, error) {
 	d := net.Dialer{Timeout: dialTimeout}
 	nc, err := d.DialContext(ctx, "tcp", addr)
 	if err != nil {
-		return nil, errors.Wrap(err, "Dial")
+		return nil, errors.Wrap(err, "dialAndNegotiate")
 	}
 	c := &Conn{nc: nc, reqTO: DefaultRequestTimeout}
 	deadline := time.Now().Add(dialTimeout)
 	if err := nc.SetDeadline(deadline); err != nil {
 		_ = nc.Close()
-		return nil, errors.Wrap(err, "Dial")
+		return nil, errors.Wrap(err, "dialAndNegotiate")
 	}
-	if err := c.handshake(username, password); err != nil {
+	if err := c.apiVersions(); err != nil {
 		_ = nc.Close()
-		return nil, errors.Wrap(err, "Dial")
+		return nil, errors.Wrap(err, "dialAndNegotiate")
 	}
-	if err := nc.SetDeadline(time.Time{}); err != nil {
+	if err := c.saslHandshake(mechanism); err != nil {
 		_ = nc.Close()
-		return nil, errors.Wrap(err, "Dial")
+		return nil, errors.Wrap(err, "dialAndNegotiate")
 	}
 	return c, nil
 }
@@ -159,56 +203,90 @@ func (c *Conn) RoundTripRequest(req kmsg.Request, clientID string) ([]byte, erro
 	return respFrame[off:], nil
 }
 
-// handshake performs ApiVersions then SaslHandshake then SaslAuthenticate
-// against the upstream broker using the supplied PLAIN credentials.
-func (c *Conn) handshake(username, password string) error {
-	// 1. ApiVersions v0 (smallest, broadest compat).
+// apiVersions runs an ApiVersions v0 exchange against the upstream broker.
+// v0 is the smallest, broadest-compat variant.
+func (c *Conn) apiVersions() error {
 	avReq := kmsg.NewPtrApiVersionsRequest()
 	avReq.SetVersion(0)
 	if _, err := c.directRoundTrip(avReq, protocol.ApiVersionsKey, 0); err != nil {
-		return errors.Wrap(err, "handshake")
+		return errors.Wrap(err, "apiVersions")
 	}
+	return nil
+}
 
-	// 2. SaslHandshake v1 — selects mechanism PLAIN.
+// saslHandshake runs SaslHandshake v1 selecting the given mechanism.
+func (c *Conn) saslHandshake(mechanism string) error {
 	hsReq := kmsg.NewPtrSASLHandshakeRequest()
 	hsReq.SetVersion(1)
-	hsReq.Mechanism = auth.MechanismPlain
+	hsReq.Mechanism = mechanism
 	hsRespBody, err := c.directRoundTrip(hsReq, protocol.SaslHandshakeKey, 1)
 	if err != nil {
-		return errors.Wrap(err, "handshake")
+		return errors.Wrap(err, "saslHandshake")
 	}
 	hsResp := kmsg.NewPtrSASLHandshakeResponse()
 	hsResp.SetVersion(1)
 	if err := hsResp.ReadFrom(hsRespBody); err != nil {
-		return errors.Wrap(err, "handshake")
+		return errors.Wrap(err, "saslHandshake")
 	}
 	if hsResp.ErrorCode != 0 {
-		return errors.Errorf("handshake: upstream SaslHandshake error code %d", hsResp.ErrorCode)
+		return errors.Errorf("saslHandshake: upstream error code %d", hsResp.ErrorCode)
 	}
+	return nil
+}
 
-	// 3. SaslAuthenticate v1 — forward the client's PLAIN credentials.
+// plainAuthenticate completes a single-shot SASL/PLAIN authenticate against
+// the upstream broker using the supplied credentials.
+func (c *Conn) plainAuthenticate(username, password string) error {
 	authReq := kmsg.NewPtrSASLAuthenticateRequest()
 	authReq.SetVersion(1)
 	authReq.SASLAuthBytes = []byte("\x00" + username + "\x00" + password)
 	authRespBody, err := c.directRoundTrip(authReq, protocol.SaslAuthenticateKey, 1)
 	if err != nil {
-		return errors.Wrap(err, "handshake")
+		return errors.Wrap(err, "plainAuthenticate")
 	}
 	authResp := kmsg.NewPtrSASLAuthenticateResponse()
 	authResp.SetVersion(1)
 	if err := authResp.ReadFrom(authRespBody); err != nil {
-		return errors.Wrap(err, "handshake")
+		return errors.Wrap(err, "plainAuthenticate")
 	}
 	if authResp.ErrorCode != 0 {
 		msg := ""
 		if authResp.ErrorMessage != nil {
 			msg = *authResp.ErrorMessage
 		}
-		return errors.Errorf("handshake: upstream SaslAuthenticate error %d: %s", authResp.ErrorCode, msg)
+		return errors.Errorf("plainAuthenticate: upstream error %d: %s", authResp.ErrorCode, msg)
 	}
 	return nil
 }
 
+// RelaySASLAuthenticate forwards a single SCRAM SaslAuthenticate payload
+// (the opaque SASLAuthBytes blob produced by the client) to the upstream
+// broker and returns the upstream's response payload, error code, and
+// optional error message. err is non-nil only on transport / decoding
+// failures; protocol-level errors are surfaced via errCode/errMsg so the
+// caller can relay them to the client unchanged.
+func (c *Conn) RelaySASLAuthenticate(payload []byte) (respPayload []byte, errCode int16, errMsg string, err error) {
+	req := kmsg.NewPtrSASLAuthenticateRequest()
+	req.SetVersion(1)
+	req.SASLAuthBytes = payload
+	if err := c.applyRequestDeadline(); err != nil {
+		return nil, 0, "", errors.Wrap(err, "RelaySASLAuthenticate")
+	}
+	body, err := c.directRoundTrip(req, protocol.SaslAuthenticateKey, 1)
+	if err != nil {
+		return nil, 0, "", errors.Wrap(err, "RelaySASLAuthenticate")
+	}
+	resp := kmsg.NewPtrSASLAuthenticateResponse()
+	resp.SetVersion(1)
+	if err := resp.ReadFrom(body); err != nil {
+		return nil, 0, "", errors.Wrap(err, "RelaySASLAuthenticate")
+	}
+	if resp.ErrorMessage != nil {
+		errMsg = *resp.ErrorMessage
+	}
+	return resp.SASLAuthBytes, resp.ErrorCode, errMsg, nil
+}
+
 // directRoundTrip is used only during the handshake when we control both
 // sides of the framing and don't need correlation-id rewriting against an
 // outer client. It returns the response body slice positioned after the
diff --git a/upstream/conn_test.go b/upstream/conn_test.go
new file mode 100644
index 0000000..79bd782
--- /dev/null
+++ b/upstream/conn_test.go
@@ -0,0 +1,54 @@
+package upstream
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestRelaySASLAuthenticate_AppliesRequestDeadline verifies the SCRAM
+// relay path applies a per-request deadline so a silent upstream cannot
+// block the caller indefinitely. Regression test for a Copilot review
+// finding on the SCRAM relay PR.
+func TestRelaySASLAuthenticate_AppliesRequestDeadline(t *testing.T) {
+	t.Parallel()
+
+	// Listener that accepts connections but never reads or writes.
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	defer func() { _ = ln.Close() }()
+
+	accepted := make(chan net.Conn, 1)
+	go func() {
+		nc, aerr := ln.Accept()
+		if aerr != nil {
+			return
+		}
+		accepted <- nc
+	}()
+
+	nc, err := net.Dial("tcp", ln.Addr().String())
+	require.NoError(t, err)
+	defer func() { _ = nc.Close() }()
+
+	// Drain the server-side accepted conn so the goroutine doesn't leak.
+	defer func() {
+		select {
+		case sc := <-accepted:
+			_ = sc.Close()
+		case <-time.After(time.Second):
+		}
+	}()
+
+	c := &Conn{nc: nc, reqTO: 100 * time.Millisecond}
+
+	start := time.Now()
+	_, _, _, err = c.RelaySASLAuthenticate([]byte("client-first-message-bare"))
+	elapsed := time.Since(start)
+
+	require.Error(t, err, "expected deadline error from silent upstream")
+	assert.Less(t, elapsed, 2*time.Second, "RelaySASLAuthenticate hung past deadline")
+}