From 7da01385014d6f1cec643adce6d808ef5ea1b253 Mon Sep 17 00:00:00 2001
From: eescobar <eescobar@twilio.com>
Date: Thu, 21 Aug 2025 16:20:48 -0500
Subject: [PATCH 1/3] [AUTHR-376] Add oauthbearer support

---
 sasl/oauthbearer/oauthbearer.go | 60 +++++++++++++++++++++++++++++++++
 1 file changed, 60 insertions(+)
 create mode 100644 sasl/oauthbearer/oauthbearer.go

diff --git a/sasl/oauthbearer/oauthbearer.go b/sasl/oauthbearer/oauthbearer.go
new file mode 100644
index 000000000..f4f4ab059
--- /dev/null
+++ b/sasl/oauthbearer/oauthbearer.go
@@ -0,0 +1,60 @@
+package plain
+
+import (
+	"context"
+	"errors"
+	"sort"
+
+	"github.com/segmentio/kafka-go/sasl"
+)
+
+// Mechanism implements the AUTHBEARER mechanism and passes the token
+type Mechanism struct {
+	Zid        string
+	Token      string
+	Extensions map[string]string
+}
+
+func (Mechanism) Name() string {
+	return "OAUTHBEARER"
+}
+
+func (m Mechanism) Start(ctx context.Context) (sasl.StateMachine, []byte, error) {
+	type kv struct {
+		k string
+		v string
+	}
+	kvs := make([]kv, 0, len(m.Extensions))
+	for k, v := range m.Extensions {
+		if len(k) == 0 {
+			continue
+		}
+		kvs = append(kvs, kv{k, v})
+	}
+	sort.Slice(kvs, func(i, j int) bool { return kvs[i].k < kvs[j].k })
+
+	gs2Header := "n,"
+	if m.Zid != "" {
+		gs2Header += "a=" + m.Zid
+	}
+	gs2Header += ","
+	init := []byte(gs2Header + "\x01auth=Bearer ")
+	init = append(init, m.Token...)
+	init = append(init, '\x01')
+	for _, kv := range kvs {
+		init = append(init, kv.k...)
+		init = append(init, '=')
+		init = append(init, kv.v...)
+		init = append(init, '\x01')
+	}
+	init = append(init, '\x01')
+
+	return m, init, nil
+}
+
+func (m Mechanism) Next(ctx context.Context, challenge []byte) (bool, []byte, error) {
+	if len(challenge) != 0 {
+		return false, nil, errors.New("unexpected data in oauth response")
+	}
+	return true, nil, nil
+}

From ed43b00a9199ed8ff0c51cdf0f2adfe5788744e2 Mon Sep 17 00:00:00 2001
From: eescobar <eescobar@twilio.com>
Date: Thu, 21 Aug 2025 16:36:51 -0500
Subject: [PATCH 2/3] [AUTHR-376] Fix typo

---
 sasl/oauthbearer/oauthbearer.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sasl/oauthbearer/oauthbearer.go b/sasl/oauthbearer/oauthbearer.go
index f4f4ab059..c603be83e 100644
--- a/sasl/oauthbearer/oauthbearer.go
+++ b/sasl/oauthbearer/oauthbearer.go
@@ -8,7 +8,7 @@ import (
 	"github.com/segmentio/kafka-go/sasl"
 )
 
-// Mechanism implements the AUTHBEARER mechanism and passes the token
+// Mechanism implements the OAUTHBEARER mechanism and passes the token
 type Mechanism struct {
 	Zid        string
 	Token      string

From 0fd6edfe4baf88905355313a140c85656b2eb54f Mon Sep 17 00:00:00 2001
From: eescobar <eescobar@twilio.com>
Date: Thu, 21 Aug 2025 17:11:38 -0500
Subject: [PATCH 3/3] [AUTHR-376] Fix typo

---
 sasl/oauthbearer/oauthbearer.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/sasl/oauthbearer/oauthbearer.go b/sasl/oauthbearer/oauthbearer.go
index c603be83e..fae66b661 100644
--- a/sasl/oauthbearer/oauthbearer.go
+++ b/sasl/oauthbearer/oauthbearer.go
@@ -1,4 +1,4 @@
-package plain
+package oauthbearer
 
 import (
 	"context"
@@ -8,7 +8,7 @@ import (
 	"github.com/segmentio/kafka-go/sasl"
 )
 
-// Mechanism implements the OAUTHBEARER mechanism and passes the token
+// Mechanism implements the OAUTHBEARER mechanism and passes the token.
 type Mechanism struct {
 	Zid        string
 	Token      string