rpk: enable OAUTHBEARER for remote debug bundle

src/go/rpk/go.mod

@@ -46,7 +46,7 @@ require (
 	github.com/prometheus/client_model v0.6.2
 	github.com/prometheus/common v0.67.5
 	github.com/redpanda-data/common-go/proto v0.0.0-20260223115805-73fb9bd9c2c0
-	github.com/redpanda-data/common-go/rpadmin v0.2.5
+	github.com/redpanda-data/common-go/rpadmin v0.2.6
 	github.com/redpanda-data/common-go/rpsr v0.1.4
 	github.com/redpanda-data/protoc-gen-go-mcp v0.0.0-20250930092048-a98b94b5957a
 	github.com/rs/xid v1.6.0

src/go/rpk/go.sum

@@ -279,8 +279,8 @@ github.com/redpanda-data/common-go/net v0.1.0 h1:JnJioRJuL961r1QXiJQ1tW9+yEaJfu8
 github.com/redpanda-data/common-go/net v0.1.0/go.mod h1:iOdNkjxM7a1T8F3cYHTaKIPFCHzzp/ia6TN+Z+7Tt5w=
 github.com/redpanda-data/common-go/proto v0.0.0-20260223115805-73fb9bd9c2c0 h1:HgdM3npLEzqKyrzni05JwFb9B/Yd7T7hdFYgaUY7d38=
 github.com/redpanda-data/common-go/proto v0.0.0-20260223115805-73fb9bd9c2c0/go.mod h1:4TOyhdEvR/hlk0RjHXzmO0hYmWBpGGI1qiFPNv211O4=
-github.com/redpanda-data/common-go/rpadmin v0.2.5 h1:mKYr5ffO6TmBwNSAHgnMI703bbf8961GczTDUFxvkqA=
-github.com/redpanda-data/common-go/rpadmin v0.2.5/go.mod h1:uOAY10WXPtcDPU0aUdpkqHR+b1BqUvRhlvMf0vha73A=
+github.com/redpanda-data/common-go/rpadmin v0.2.6 h1:fVfWvyw1xJy7lWBQvyEdLxx40Pkq74DuBwOdCKS3lZ0=
+github.com/redpanda-data/common-go/rpadmin v0.2.6/go.mod h1:uOAY10WXPtcDPU0aUdpkqHR+b1BqUvRhlvMf0vha73A=
 github.com/redpanda-data/common-go/rpsr v0.1.4 h1:d9lu5q5wyhZWBYR1GnZkq+eZGKU0qoaSwwybRS9Uk2k=
 github.com/redpanda-data/common-go/rpsr v0.1.4/go.mod h1:qVa7b0yaCRdZDn5dcZ9CazqVr4jYbgtOJUywI2X3G3I=
 github.com/redpanda-data/protoc-gen-go-mcp v0.0.0-20250930092048-a98b94b5957a h1:jNHT6Fcy/rBAFnX8rjbwrJ+lSF2Ufa1jqmnCV6m6RKY=

src/go/rpk/pkg/cli/debug/remotebundle/start.go

@@ -78,16 +78,6 @@ Use the flag '--no-confirm' to avoid the confirmation prompt.
 			out.MaybeDie(err, "rpk unable to load config: %v", err)
 			config.CheckExitCloudAdmin(p)
 
-			// Remote debug bundle forwards SASL credentials to the broker so
-			// rpk running on the broker can authenticate to Kafka. The broker
-			// API currently only accepts a SCRAM-shaped payload; until that
-			// gains OAUTHBEARER support (tracked in redpanda#30222), reject
-			// up front rather than silently sending the request with no auth
-			// (which then fails confusingly on secured clusters).
-			if p.KafkaAPI.SASL != nil && strings.EqualFold(p.KafkaAPI.SASL.Mechanism, adminapi.OAuthBearer) {
-				out.Die("OAUTHBEARER is not yet supported for remote debug bundle collection; use SCRAM-SHA-256 or SCRAM-SHA-512 credentials to run this command")
-			}
-
 			if !noConfirm {
 				printBrokers(p.AdminAPI.Addresses)
 				confirmed, err := out.Confirm("Confirm debug bundle collection from these brokers?")
@@ -260,9 +250,12 @@ func (o *remoteBundleOptions) toRpadminOptions(p *config.RpkProfile) []rpadmin.D
 		}
 		opts = append(opts, rpadmin.WithLabelSelector(dbls))
 	}
-	if p.HasSASLCredentials() {
-		s := p.KafkaAPI.SASL
-		opts = append(opts, rpadmin.WithSCRAMAuthentication(s.User, s.Password, s.Mechanism))
+	if s := p.KafkaAPI.SASL; s != nil {
+		if strings.EqualFold(s.Mechanism, adminapi.OAuthBearer) {
+			opts = append(opts, rpadmin.WithOAuthBearerAuthentication(adminapi.OAuthBearerToken(s.Password)))
+		} else if p.HasSASLCredentials() {
+			opts = append(opts, rpadmin.WithSCRAMAuthentication(s.User, s.Password, s.Mechanism))
+		}
 	}
 	if tls := p.KafkaAPI.TLS; tls != nil {
 		opts = append(opts, rpadmin.WithTLS(true, tls.InsecureSkipVerify))

tests/rptest/services/admin.py

@@ -26,7 +26,7 @@
 from requests.exceptions import HTTPError, RequestException
 from urllib3.util.retry import Retry
 
-from rptest.services.redpanda_types import SaslCredentials
+from rptest.services.redpanda_types import OAuthBearerCredentials, SaslCredentials
 from rptest.util import not_none, wait_until_result
 from rptest.utils.mode_checks import is_debug_mode
 
@@ -456,7 +456,7 @@ class DebugBundleLabelSelection:
 
 
 class DebugBundleStartConfigParams(NamedTuple):
-    authentication: Optional[SaslCredentials] = None
+    authentication: Optional[SaslCredentials | OAuthBearerCredentials] = None
     controller_logs_size_limit_bytes: Optional[int] = None
     cpu_profiler_wait_seconds: Optional[int] = None
     logs_since: Optional[str] = None

tests/rptest/services/redpanda_types.py

@@ -46,6 +46,14 @@ def mechanism(self):
         return self.algorithm
 
 
+@dataclass
+class OAuthBearerCredentials:
+    """Bearer token credentials for OAUTHBEARER authentication in debug bundle."""
+
+    token: str
+    mechanism: str = "OAUTHBEARER"
+
+
 class SecurityProtocol(Enum):
     """The four possible security protocol options for Kafka authentication."""
 

tests/rptest/tests/debug_bundle_test.py

@@ -15,13 +15,15 @@
 import time
 import zipfile
 from typing import Optional
+from urllib.parse import urlparse
 from uuid import UUID, uuid4
 
 import requests
 from ducktape.cluster.cluster import ClusterNode
 from ducktape.mark import matrix
 from ducktape.services.service import Service
 from ducktape.utils.util import wait_until
+from keycloak import KeycloakOpenID
 
 from rptest.clients.rpk import RpkTool
 from rptest.services.admin import (
@@ -31,6 +33,7 @@
     DebugBundleStartConfigParams,
 )
 from rptest.services.cluster import cluster
+from rptest.services.keycloak import DEFAULT_REALM, KeycloakService
 from rptest.services.redpanda import (
     LoggingConfig,
     MetricSamples,
@@ -39,11 +42,13 @@
     SecurityConfig,
     TLSProvider,
 )
-from rptest.services.redpanda_types import SaslCredentials
+from rptest.services.redpanda_types import OAuthBearerCredentials, SaslCredentials
 from rptest.services.tls import Certificate, CertificateAuthority, TLSCertManager
 from rptest.tests.redpanda_test import RedpandaTest
 from rptest.util import wait_until_result
 
+_OIDC_TOKEN_AUDIENCE = "account"
+
 
 class DebugBundleErrorCode:
     SUCCESS = 0
@@ -734,3 +739,98 @@ def test_validate_tls(self):
         )
         content = self._retrieve_file(node=node)
         self._validate_topic_name_in_response(content, self.topic_name, True)
+
+
+class DebugBundleOAuthBearerAuthn(DebugBundleTestBase):
+    """
+    End-to-end test verifying OAUTHBEARER authentication for the debug bundle.
+
+    The broker receives {mechanism: OAUTHBEARER, token: JWT} in the start
+    request and forwards it as -Xsasl.mechanism=OAUTHBEARER and
+    -Xpass=token:<JWT> to the rpk subprocess. The subprocess uses those
+    credentials to authenticate to Kafka, producing a successful bundle.
+    """
+
+    _CLIENT_ID = "debug-bundle-test"
+    _TOPIC = "oauthbearer_test_topic"
+
+    def __init__(self, context):
+        # Allocate the Keycloak node now so we can derive its hostname for
+        # oidc_discovery_url before creating the Redpanda service.  The node
+        # is not started until setUp() so that self.logger is available for
+        # error handling by then.
+        self.keycloak = KeycloakService(context)
+        kc_node = self.keycloak.nodes[0]
+        kc_discovery_url = self.keycloak.get_discovery_url(kc_node)
+
+        security = SecurityConfig()
+        security.enable_sasl = True
+        security.sasl_mechanisms = ["SCRAM", "OAUTHBEARER"]
+        security.kafka_enable_authorization = False
+        security.endpoint_authn_method = "sasl"
+
+        super().__init__(
+            context,
+            num_brokers=1,
+            security=security,
+            extra_rp_conf={
+                "oidc_discovery_url": kc_discovery_url,
+                "oidc_token_audience": _OIDC_TOKEN_AUDIENCE,
+            },
+        )
+
+        self.super_admin = Admin(self.redpanda)
+        self.rpk = RpkTool(
+            self.redpanda,
+            username=self.redpanda.SUPERUSER_CREDENTIALS.username,
+            password=self.redpanda.SUPERUSER_CREDENTIALS.password,
+            sasl_mechanism=self.redpanda.SUPERUSER_CREDENTIALS.algorithm,
+        )
+
+    def setUp(self):
+        kc_node = self.keycloak.nodes[0]
+        try:
+            self.keycloak.start_node(kc_node)
+        except Exception as e:
+            self.logger.error(f"Keycloak failed to start: {e}")
+            self.keycloak.clean_node(kc_node)
+            raise
+
+        super().setUp()
+
+        self.keycloak.admin.create_client(self._CLIENT_ID)
+        self.oauth_cfg = self.keycloak.generate_oauth_config(kc_node, self._CLIENT_ID)
+        self.rpk.create_topic(self._TOPIC)
+
+    def _get_access_token(self) -> str:
+        cfg = self.oauth_cfg
+        parsed = urlparse(cfg.token_endpoint)
+        openid = KeycloakOpenID(
+            server_url=f"{parsed.scheme}://{parsed.netloc}",
+            client_id=cfg.client_id,
+            client_secret_key=cfg.client_secret,
+            realm_name=DEFAULT_REALM,
+            verify=False,
+        )
+        return openid.token(grant_type="client_credentials")["access_token"]
+
+    @cluster(num_nodes=2)
+    def test_oauthbearer_auth(self):
+        """
+        Verify that the broker correctly forwards the bearer token to the rpk
+        subprocess, which uses OAUTHBEARER to authenticate to Kafka and produce
+        a successful debug bundle containing the expected topic metadata.
+        """
+        node = self.redpanda.started_nodes()[0]
+        token = self._get_access_token()
+
+        job_id = uuid4()
+        config = DebugBundleStartConfigParams(
+            authentication=OAuthBearerCredentials(token=token)
+        )
+        self._run_debug_bundle(
+            job_id=job_id, node=node, config=config, admin=self.super_admin
+        )
+
+        content = self._retrieve_file(node=node, admin=self.super_admin)
+        self._validate_topic_name_in_response(content, self._TOPIC, True)