rpk: add OAUTHBEARER SASL mechanism support

src/go/rpk/pkg/adminapi/admin.go

@@ -40,11 +40,18 @@ const (
 	ScramSha256 = "SCRAM-SHA-256"
 	ScramSha512 = "SCRAM-SHA-512"
 	CloudOIDC   = "CLOUD-OIDC"
+	OAuthBearer = "OAUTHBEARER"
 )
 
 // GetAuth gets the rpadmin.Auth from the rpk profile.
 func GetAuth(p *config.RpkProfile) (rpadmin.Auth, error) {
 	switch {
+	case p.KafkaAPI.SASL != nil && strings.EqualFold(p.KafkaAPI.SASL.Mechanism, OAuthBearer):
+		token := oauthBearerToken(p.KafkaAPI.SASL.Password)
+		if token == "" {
+			return nil, errors.New("OAUTHBEARER requires a token passed via --sasl-password")
+		}
+		return &rpadmin.BearerToken{Token: token}, nil
 	case p.KafkaAPI.SASL != nil && p.KafkaAPI.SASL.Mechanism != CloudOIDC:
 		return &rpadmin.BasicAuth{Username: p.KafkaAPI.SASL.User, Password: p.KafkaAPI.SASL.Password}, nil
 	case p.KafkaAPI.SASL != nil && p.KafkaAPI.SASL.Mechanism == CloudOIDC:
@@ -72,6 +79,15 @@ func GetAuth(p *config.RpkProfile) (rpadmin.Auth, error) {
 	}
 }
 
+// oauthBearerToken extracts the bearer token from the SASL password field.
+// It accepts both "token:<TOKEN>" format and a raw token string.
+func oauthBearerToken(password string) string {
+	if t, ok := strings.CutPrefix(password, "token:"); ok {
+		return t
+	}
+	return password
+}
+
 // NewClient returns an rpadmin.AdminAPI client that talks to each of the
 // addresses in the rpk.admin_api section of the config.
 func NewClient(ctx context.Context, fs afero.Fs, p *config.RpkProfile, opts ...rpadmin.Opt) (*rpadmin.AdminAPI, error) {

src/go/rpk/pkg/adminapi/auth_test.go

@@ -0,0 +1,143 @@
+package adminapi
+
+import (
+	"testing"
+
+	"github.com/redpanda-data/common-go/rpadmin"
+	"github.com/redpanda-data/redpanda/src/go/rpk/pkg/config"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGetAuth(t *testing.T) {
+	tests := []struct {
+		name    string
+		profile *config.RpkProfile
+		wantTyp rpadmin.Auth
+		wantErr string
+	}{
+		{
+			name:    "no SASL returns NopAuth",
+			profile: &config.RpkProfile{},
+			wantTyp: &rpadmin.NopAuth{},
+		},
+		{
+			name: "SCRAM-SHA-256 returns BasicAuth",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						User:      "admin",
+						Password:  "secret",
+						Mechanism: "SCRAM-SHA-256",
+					},
+				},
+			},
+			wantTyp: &rpadmin.BasicAuth{Username: "admin", Password: "secret"},
+		},
+		{
+			name: "OAUTHBEARER with token prefix returns BearerToken",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						Password:  "token:my-jwt-token",
+						Mechanism: "OAUTHBEARER",
+					},
+				},
+			},
+			wantTyp: &rpadmin.BearerToken{Token: "my-jwt-token"},
+		},
+		{
+			name: "OAUTHBEARER with raw token returns BearerToken",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						Password:  "my-jwt-token",
+						Mechanism: "OAUTHBEARER",
+					},
+				},
+			},
+			wantTyp: &rpadmin.BearerToken{Token: "my-jwt-token"},
+		},
+		{
+			name: "OAUTHBEARER case-insensitive",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						Password:  "my-jwt-token",
+						Mechanism: "oauthbearer",
+					},
+				},
+			},
+			wantTyp: &rpadmin.BearerToken{Token: "my-jwt-token"},
+		},
+		{
+			name: "OAUTHBEARER with empty password errors",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						Mechanism: "OAUTHBEARER",
+					},
+				},
+			},
+			wantErr: "OAUTHBEARER requires a token",
+		},
+		{
+			name: "OAUTHBEARER with token: prefix only errors",
+			profile: &config.RpkProfile{
+				KafkaAPI: config.RpkKafkaAPI{
+					SASL: &config.SASL{
+						Password:  "token:",
+						Mechanism: "OAUTHBEARER",
+					},
+				},
+			},
+			wantErr: "OAUTHBEARER requires a token",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := GetAuth(tt.profile)
+			if tt.wantErr != "" {
+				require.ErrorContains(t, err, tt.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			require.IsType(t, tt.wantTyp, got)
+			require.Equal(t, tt.wantTyp, got)
+		})
+	}
+}
+
+func Test_oauthBearerToken(t *testing.T) {
+	tests := []struct {
+		name     string
+		password string
+		want     string
+	}{
+		{
+			name:     "token prefix stripped",
+			password: "token:eyJhbGciOiJSUzI1NiJ9.payload.sig",
+			want:     "eyJhbGciOiJSUzI1NiJ9.payload.sig",
+		},
+		{
+			name:     "raw token returned as-is",
+			password: "eyJhbGciOiJSUzI1NiJ9.payload.sig",
+			want:     "eyJhbGciOiJSUzI1NiJ9.payload.sig",
+		},
+		{
+			name:     "empty password returns empty",
+			password: "",
+			want:     "",
+		},
+		{
+			name:     "token prefix only returns empty",
+			password: "token:",
+			want:     "",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := oauthBearerToken(tt.password)
+			require.Equal(t, tt.want, got)
+		})
+	}
+}

src/go/rpk/pkg/config/params.go

@@ -871,7 +871,7 @@ func (p *Params) InstallSASLFlags(cmd *cobra.Command) {
 
 	pf.StringVar(&p.user, FlagSASLUser, "", "SASL user to be used for authentication")
 	pf.StringVar(&p.password, "password", "", "SASL password to be used for authentication")
-	pf.StringVar(&p.saslMechanism, "sasl-mechanism", "", "The authentication mechanism to use (SCRAM-SHA-256, SCRAM-SHA-512)")
+	pf.StringVar(&p.saslMechanism, "sasl-mechanism", "", "The authentication mechanism to use (SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER)")
 
 	pf.MarkHidden(FlagSASLUser)
 	pf.MarkHidden("password")

src/go/rpk/pkg/config/profile_doc.go

@@ -35,7 +35,7 @@ var profileFieldDocs = map[string]string{
 	"kafka_api.sasl":                     "SASL authentication configuration",
 	"kafka_api.sasl.user":                "username for authentication",
 	"kafka_api.sasl.password":            "password for authentication",
-	"kafka_api.sasl.mechanism":           "SCRAM-SHA-256 or SCRAM-SHA-512",
+	"kafka_api.sasl.mechanism":           "SCRAM-SHA-256, SCRAM-SHA-512, or OAUTHBEARER",
 
 	"admin_api":                          "Admin API connection configuration",
 	"admin_api.addresses":                "Comma-separated list of Admin API addresses (host:port)",

src/go/rpk/pkg/kafka/client_franz.go

@@ -121,22 +121,28 @@ func NewFranzClient(fs afero.Fs, p *config.RpkProfile, extraOpts ...kgo.Opt) (*k
 				Token: a.AuthToken,
 			}).AsMechanism()))
 		} else {
-			a := scram.Auth{
-				User: k.SASL.User,
-				Pass: k.SASL.Password,
-			}
 			switch name := strings.ToUpper(k.SASL.Mechanism); name {
 			case "SCRAM-SHA-256", "": // we default to SCRAM-SHA-256 -- people commonly specify user & pass without --sasl-mechanism
+				a := scram.Auth{User: k.SASL.User, Pass: k.SASL.Password}
 				opts = append(opts, kgo.SASL(a.AsSha256Mechanism()))
 			case "SCRAM-SHA-512":
+				a := scram.Auth{User: k.SASL.User, Pass: k.SASL.Password}
 				opts = append(opts, kgo.SASL(a.AsSha512Mechanism()))
 			case "PLAIN":
 				opts = append(opts, kgo.SASL((&plain.Auth{
 					User: k.SASL.User,
 					Pass: k.SASL.Password,
 				}).AsMechanism()))
+			case "OAUTHBEARER":
+				token := oauthBearerToken(k.SASL.Password)
+				if token == "" {
+					return nil, fmt.Errorf("OAUTHBEARER requires a token passed via --sasl-password")
+				}
+				opts = append(opts, kgo.SASL((koauth.Auth{
+					Token: token,
+				}).AsMechanism()))
 			default:
-				return nil, fmt.Errorf("unknown SASL mechanism %q, supported: [SCRAM-SHA-256, SCRAM-SHA-512, PLAIN]", name)
+				return nil, fmt.Errorf("unknown SASL mechanism %q, supported: [SCRAM-SHA-256, SCRAM-SHA-512, PLAIN, OAUTHBEARER]", name)
 			}
 		}
 	}
@@ -154,6 +160,15 @@ func NewFranzClient(fs afero.Fs, p *config.RpkProfile, extraOpts ...kgo.Opt) (*k
 	return kgo.NewClient(opts...)
 }
 
+// oauthBearerToken extracts the bearer token from the SASL password field.
+// It accepts both "token:<TOKEN>" format and a raw token string.
+func oauthBearerToken(password string) string {
+	if t, ok := strings.CutPrefix(password, "token:"); ok {
+		return t
+	}
+	return password
+}
+
 // NewAdmin returns a franz-go admin client.
 func NewAdmin(fs afero.Fs, p *config.RpkProfile, extraOpts ...kgo.Opt) (*kadm.Client, error) {
 	cl, err := NewFranzClient(fs, p, extraOpts...)

src/go/rpk/pkg/kafka/client_franz_test.go

@@ -0,0 +1,45 @@
+package kafka
+
+import "testing"
+
+func Test_oauthBearerToken(t *testing.T) {
+	tests := []struct {
+		name     string
+		password string
+		want     string
+	}{
+		{
+			name:     "token prefix stripped",
+			password: "token:eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test",
+			want:     "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test",
+		},
+		{
+			name:     "raw token returned as-is",
+			password: "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test",
+			want:     "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.test",
+		},
+		{
+			name:     "empty password returns empty",
+			password: "",
+			want:     "",
+		},
+		{
+			name:     "token prefix only returns empty",
+			password: "token:",
+			want:     "",
+		},
+		{
+			name:     "token prefix is case-sensitive",
+			password: "Token:abc",
+			want:     "Token:abc",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := oauthBearerToken(tt.password)
+			if got != tt.want {
+				t.Errorf("oauthBearerToken(%q) = %q, want %q", tt.password, got, tt.want)
+			}
+		})
+	}
+}

src/go/rpk/pkg/schemaregistry/client.go

@@ -61,6 +61,12 @@ func NewClient(fs afero.Fs, p *config.RpkProfile) (*rpsr.Client, error) {
 	}
 
 	switch {
+	case p.KafkaAPI.SASL != nil && strings.EqualFold(p.KafkaAPI.SASL.Mechanism, adminapi.OAuthBearer):
+		token := oauthBearerToken(p.KafkaAPI.SASL.Password)
+		if token == "" {
+			return nil, errors.New("OAUTHBEARER requires a token passed via --sasl-password")
+		}
+		opts = append(opts, sr.BearerToken(token))
 	case p.HasSASLCredentials() && p.KafkaAPI.SASL.Mechanism != adminapi.CloudOIDC:
 		opts = append(opts, sr.BasicAuth(p.KafkaAPI.SASL.User, p.KafkaAPI.SASL.Password))
 	case p.KafkaAPI.SASL != nil && p.KafkaAPI.SASL.Mechanism == adminapi.CloudOIDC:
@@ -93,6 +99,15 @@ func NewClient(fs afero.Fs, p *config.RpkProfile) (*rpsr.Client, error) {
 	return rpsr.NewClient(srCl)
 }
 
+// oauthBearerToken extracts the bearer token from the SASL password field.
+// It accepts both "token:<TOKEN>" format and a raw token string.
+func oauthBearerToken(password string) string {
+	if t, ok := strings.CutPrefix(password, "token:"); ok {
+		return t
+	}
+	return password
+}
+
 // IsSoftDeleteError checks whether the error is a SoftDeleteError. This error
 // occurs when attempting to soft-delete a schema that was already marked as
 // soft deleted.