debug_bundle: support OAUTHBEARER auth in broker-side admin API

src/v/debug_bundle/debug_bundle_service.cc

@@ -600,6 +600,14 @@ result<std::vector<ss::sstring>> service::build_rpk_arguments(
               rv.emplace_back(
                 ssx::sformat(
                   "{}={}", sasl_mechanism_variable, creds.mechanism));
+          },
+          [&rv](const bearer_creds& creds) mutable {
+              // rpk accepts -Xpass=token:<TOKEN> for OAUTHBEARER
+              rv.emplace_back(
+                ssx::sformat("{}=token:{}", password_variable, creds.token));
+              rv.emplace_back(
+                ssx::sformat(
+                  "{}={}", sasl_mechanism_variable, creds.mechanism));
           });
     }
     if (params.controller_logs_size_limit_bytes.has_value()) {

src/v/debug_bundle/json.h

@@ -167,12 +167,46 @@ debug_bundle::result<T> from_json(const json::Value& v) {
             return std::move(sc);
         }
         return parse_error(": expected scram_creds");
+    } else if constexpr (std::is_same_v<T, bearer_creds>) {
+        if (v.IsObject()) {
+            auto o = v.GetObject();
+            bearer_creds bc;
+            if (
+              auto r = from_json<decltype(bc.token)>(o, "token", true);
+              r.has_value()) {
+                bc.token = std::move(r).assume_value();
+            } else {
+                return std::move(r).assume_error();
+            }
+            if (
+              auto r = from_json<decltype(bc.mechanism)>(o, "mechanism", true);
+              r.has_value()) {
+                bc.mechanism = std::move(r).assume_value();
+            } else {
+                return std::move(r).assume_error();
+            }
+            return std::move(bc);
+        }
+        return parse_error(": expected bearer_creds");
     } else if constexpr (std::is_same_v<T, debug_bundle_authn_options>) {
-        auto r = from_json<scram_creds>(v);
-        if (r.has_value()) {
-            return T{std::move(r).assume_value()};
+        // Dispatch on the presence of "token" (OAUTHBEARER) vs "username"
+        // (SCRAM)
+        if (v.IsObject()) {
+            auto o = v.GetObject();
+            if (o.HasMember("token")) {
+                auto r = from_json<bearer_creds>(v);
+                if (r.has_value()) {
+                    return T{std::move(r).assume_value()};
+                }
+                return std::move(r).assume_error();
+            }
+            auto r = from_json<scram_creds>(v);
+            if (r.has_value()) {
+                return T{std::move(r).assume_value()};
+            }
+            return std::move(r).assume_error();
         }
-        return std::move(r).assume_error();
+        return parse_error(": expected authentication credentials object");
     } else if constexpr (std::is_same_v<T, partition_selection>) {
         if (v.IsString()) {
             auto r = partition_selection::from_string_view(as_string_view(v));

src/v/debug_bundle/tests/debug_bundle_service_test.cc

@@ -384,6 +384,34 @@ TEST_F_CORO(debug_bundle_service_started_fixture, test_all_parameters) {
     }
 }
 
+TEST_F_CORO(debug_bundle_service_started_fixture, test_bearer_creds_args) {
+    debug_bundle::job_id_t job_id(uuid_t::create());
+
+    ss::sstring token = "eyJhbGciOiJSUzI1NiJ9.test-payload";
+    ss::sstring mechanism = "OAUTHBEARER";
+
+    debug_bundle::debug_bundle_parameters params{
+      .authn_options = debug_bundle::bearer_creds{
+        .token = token, .mechanism = mechanism}};
+
+    co_await run_bundle(job_id, std::move(params));
+
+    auto status = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_TRUE_CORO(status.has_value()) << status.assume_error().message();
+    ASSERT_FALSE_CORO(status.assume_value().cout.empty());
+
+    // rpk-shim echoes all args; token appears as -Xpass=token:<TOKEN>
+    auto expected = ssx::sformat(
+      "debug bundle --output {}/{}.zip --verbose -Xpass=token:{} "
+      "-Xsasl.mechanism={}\n",
+      (_data_dir / debug_bundle::service::debug_bundle_dir_name).native(),
+      job_id,
+      token,
+      mechanism);
+    EXPECT_EQ(status.assume_value().cout[0], expected)
+      << status.assume_value().cout[0] << " != " << expected;
+}
+
 TEST_F_CORO(debug_bundle_service_started_fixture, try_running_multiple) {
     auto res = co_await _service.invoke_on(
       debug_bundle::service_shard, [](debug_bundle::service& s) {
@@ -674,6 +702,11 @@ ss::future<> wait_for_file_to_be_created(
     throw std::runtime_error(
       fmt::format("Timed out waiting for process file '{}' to exist", file));
 }
+
+ss::future<> wait_for_kvstore_to_populate(
+  storage::kvstore* kvstore,
+  std::chrono::seconds timeout = std::chrono::seconds{10});
+
 TEST_F_CORO(debug_bundle_service_started_fixture, check_clean_up) {
     using namespace std::chrono_literals;
     debug_bundle::job_id_t job1(uuid_t::create());
@@ -695,8 +728,12 @@ TEST_F_CORO(debug_bundle_service_started_fixture, check_clean_up) {
         ASSERT_TRUE_CORO(res.has_value()) << res.assume_error().message();
         ASSERT_NO_THROW_CORO(
           co_await wait_for_file_to_be_created(job1_file, 10s));
+        // wait_for_kvstore_to_populate is the reliable signal that set_metadata
+        // has finished: it writes the .out file then updates the kvstore.
+        // Using a 30s budget to absorb SHA256 latency on slow sandbox I/O.
         ASSERT_NO_THROW_CORO(
-          co_await wait_for_file_to_be_created(job1_out_file, 10s));
+          co_await wait_for_kvstore_to_populate(_kvstore.get(), 30s));
+        ASSERT_TRUE_CORO(co_await ss::file_exists(job1_out_file.native()));
     }
     {
         auto res
@@ -706,15 +743,15 @@ TEST_F_CORO(debug_bundle_service_started_fixture, check_clean_up) {
         ASSERT_NO_THROW_CORO(
           co_await wait_for_file_to_be_created(job2_file, 10s));
         ASSERT_NO_THROW_CORO(
-          co_await wait_for_file_to_be_created(job2_out_file, 10s));
+          co_await wait_for_kvstore_to_populate(_kvstore.get(), 30s));
+        ASSERT_TRUE_CORO(co_await ss::file_exists(job2_out_file.native()));
     }
     EXPECT_FALSE(co_await ss::file_exists(job1_file.native()));
     EXPECT_FALSE(co_await ss::file_exists(job1_out_file.native()));
 }
 
 ss::future<> wait_for_kvstore_to_populate(
-  storage::kvstore* kvstore,
-  std::chrono::seconds timeout = std::chrono::seconds{10}) {
+  storage::kvstore* kvstore, std::chrono::seconds timeout) {
     const auto start_time = debug_bundle::clock::now();
     while (debug_bundle::clock::now() - start_time <= timeout) {
         auto metadata_buf = kvstore->get(

src/v/debug_bundle/tests/json_test.cc

@@ -53,6 +53,7 @@ using JsonTestTypes = ::testing::Types<
   clock::time_point,
   time_variant,
   scram_creds,
+  bearer_creds,
   debug_bundle_authn_options,
   partition_selection,
   debug_bundle_parameters,
@@ -105,12 +106,17 @@ TYPED_TEST(JsonTypeTest, BasicType) {
           = R"({"username": "user", "password": "pass", "mechanism": "SCRAM-SHA-256"})";
         this->expected = scram_creds{
           .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}};
+    } else if constexpr (std::is_same_v<TypeParam, bearer_creds>) {
+        this->json_input
+          = R"({"token": "my-jwt-token", "mechanism": "OAUTHBEARER"})";
+        this->expected = bearer_creds{
+          .token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}};
     } else if constexpr (
       std::is_same_v<TypeParam, debug_bundle_authn_options>) {
         this->json_input
-          = R"({"username": "user", "password": "pass", "mechanism": "SCRAM-SHA-256"})";
-        this->expected = TypeParam{scram_creds{
-          .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}}};
+          = R"({"token": "my-jwt-token", "mechanism": "OAUTHBEARER"})";
+        this->expected = TypeParam{
+          bearer_creds{.token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}}};
     } else if constexpr (std::is_same_v<TypeParam, partition_selection>) {
         this->json_input = R"("foo/bar/1")";
         this->expected = {
@@ -241,11 +247,16 @@ TYPED_TEST(JsonTypeTest, TypeIsInvalid) {
         this->json_input = R"({"credential": "user:pass:SCRAM-SHA-256"})";
         this->expected = scram_creds{
           .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}};
+    } else if constexpr (std::is_same_v<TypeParam, bearer_creds>) {
+        // Missing required "token" field
+        this->json_input = R"({"mechanism": "OAUTHBEARER"})";
+        this->expected = bearer_creds{
+          .token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}};
     } else if constexpr (
       std::is_same_v<TypeParam, debug_bundle_authn_options>) {
         this->json_input = R"(42)";
-        this->expected = TypeParam{scram_creds{
-          .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}}};
+        this->expected = TypeParam{
+          bearer_creds{.token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}}};
     } else if constexpr (std::is_same_v<TypeParam, partition_selection>) {
         this->json_input = R"("invalid")";
         this->expected = {
@@ -297,12 +308,17 @@ TYPED_TEST(JsonTypeTest, ValidateControlCharacters) {
           = R"({"username": "user\r", "password": "pass", "mechanism": "SCRAM-SHA-256"})";
         this->expected = scram_creds{
           .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}};
+    } else if constexpr (std::is_same_v<TypeParam, bearer_creds>) {
+        this->json_input
+          = R"({"token": "bad\ntoken", "mechanism": "OAUTHBEARER"})";
+        this->expected = bearer_creds{
+          .token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}};
     } else if constexpr (
       std::is_same_v<TypeParam, debug_bundle_authn_options>) {
         this->json_input
-          = R"({"username": "user", "password": "\fpass", "mechanism": "SCRAM-SHA-256"})";
-        this->expected = TypeParam{scram_creds{
-          .username{"user"}, .password{"pass"}, .mechanism{"SCRAM-SHA-256"}}};
+          = R"({"token": "bad\ntoken", "mechanism": "OAUTHBEARER"})";
+        this->expected = TypeParam{
+          bearer_creds{.token{"my-jwt-token"}, .mechanism{"OAUTHBEARER"}}};
     } else {
         return;
     }
@@ -325,3 +341,48 @@ TYPED_TEST(JsonTypeTest, ValidateControlCharacters) {
       != std::string::npos)
       << res.assume_error().message();
 }
+
+/// Verify that debug_bundle_parameters accepts OAUTHBEARER authentication.
+TEST(JsonTest, ParametersWithBearerAuth) {
+    const ss::sstring json_input = R"({
+  "authentication": {
+    "token": "my-jwt-token",
+    "mechanism": "OAUTHBEARER"
+  }
+})";
+
+    json::Document doc;
+    ASSERT_NO_THROW(doc.Parse(json_input));
+    ASSERT_FALSE(doc.HasParseError());
+
+    debug_bundle::result<debug_bundle_parameters> res{outcome::success()};
+    ASSERT_NO_THROW(res = from_json<debug_bundle_parameters>(doc));
+    ASSERT_TRUE(res.has_value()) << res.assume_error().message();
+
+    const auto& params = res.assume_value();
+    ASSERT_TRUE(params.authn_options.has_value());
+    ASSERT_TRUE(
+      std::holds_alternative<bearer_creds>(params.authn_options.value()));
+    const auto& bc = std::get<bearer_creds>(params.authn_options.value());
+    EXPECT_EQ(bc.token, "my-jwt-token");
+    EXPECT_EQ(bc.mechanism, "OAUTHBEARER");
+}
+
+/// Verify that {mechanism: OAUTHBEARER} with no token is rejected with a parse
+/// error (maps to HTTP 400 in the admin API).
+TEST(JsonTest, BearerAuthMissingTokenIsRejected) {
+    const ss::sstring json_input = R"({
+  "authentication": {
+    "mechanism": "OAUTHBEARER"
+  }
+})";
+
+    json::Document doc;
+    ASSERT_NO_THROW(doc.Parse(json_input));
+    ASSERT_FALSE(doc.HasParseError());
+
+    debug_bundle::result<debug_bundle_parameters> res{outcome::success()};
+    ASSERT_NO_THROW(res = from_json<debug_bundle_parameters>(doc));
+    ASSERT_TRUE(res.has_error());
+    EXPECT_EQ(res.assume_error().code(), error_code::invalid_parameters);
+}

src/v/debug_bundle/types.h

@@ -73,8 +73,18 @@ struct scram_creds {
 
     friend bool operator==(const scram_creds&, const scram_creds&) = default;
 };
+
+/// OAUTHBEARER credentials for authn; carries the raw bearer token
+struct bearer_creds {
+    ss::sstring token;
+    /// Always "OAUTHBEARER"
+    ss::sstring mechanism;
+
+    friend bool operator==(const bearer_creds&, const bearer_creds&) = default;
+};
+
 /// Variant so it can be expanded as new authn methods are added to rpk
-using debug_bundle_authn_options = std::variant<scram_creds>;
+using debug_bundle_authn_options = std::variant<scram_creds, bearer_creds>;
 
 /// Used to collect topics and partitions for the "--partitions" option for "rpk
 /// debug_bundle"

src/v/redpanda/admin/api-doc/debug_bundle.json

@@ -28,24 +28,26 @@
               "required": [],
               "properties": {
                 "authentication": {
-                  "description": "Authentication object",
+                  "description": "Authentication credentials. Either SCRAM ({mechanism, username, password}) or OAUTHBEARER ({mechanism, token}).",
                   "type": "object",
                   "required": [
-                    "mechanism",
-                    "username",
-                    "password"
+                    "mechanism"
                   ],
                   "properties": {
                     "mechanism": {
-                      "description": "SCRAM mechanism",
+                      "description": "SASL mechanism: SCRAM-SHA-256, SCRAM-SHA-512, or OAUTHBEARER",
                       "type": "string"
                     },
                     "username": {
-                      "description": "username used by RPK to authenticate against Kafka and Admin API",
+                      "description": "Username for SCRAM authentication",
                       "type": "string"
                     },
                     "password": {
-                      "description": "password used by RPK to authenticate against Kafka and Admin API",
+                      "description": "Password for SCRAM authentication",
+                      "type": "string"
+                    },
+                    "token": {
+                      "description": "Bearer token for OAUTHBEARER authentication",
                       "type": "string"
                     }
                   }