From 5ec43ff19c1b5cf7f847e581c224db940611a3b0 Mon Sep 17 00:00:00 2001
From: Michael Boquard <michael@redpanda.com>
Date: Tue, 11 Jul 2023 22:03:54 -0400
Subject: [PATCH] k/server: Better handling of SCRAM parsing errors

Before this change, if Redpanda received a badly formatted SCRAM
message, it would not respond with an error and the offending client
would timeout with no error code.  This change will now return an error
code and print a more helpful message to the logs.

Fixes: #11349

Signed-off-by: Michael Boquard <michael@redpanda.com>
---
 src/v/kafka/server/server.cc | 38 +++++++++++++++++++++++++-----------
 1 file changed, 27 insertions(+), 11 deletions(-)

diff --git a/src/v/kafka/server/server.cc b/src/v/kafka/server/server.cc
index 8ae6402b7c26..021efbce936a 100644
--- a/src/v/kafka/server/server.cc
+++ b/src/v/kafka/server/server.cc
@@ -372,22 +372,38 @@ ss::future<response_ptr> sasl_authenticate_handler::handle(
     log_request(ctx.header(), request);
     vlog(klog.debug, "Received SASL_AUTHENTICATE {}", request);
 
-    auto result = co_await ctx.sasl()->authenticate(
-      std::move(request.data.auth_bytes));
-    if (likely(result)) {
-        sasl_authenticate_response_data data{
-          .error_code = error_code::none,
-          .error_message = std::nullopt,
-          .auth_bytes = std::move(result.value()),
-        };
-        co_return co_await ctx.respond(
-          sasl_authenticate_response(std::move(data)));
+    std::error_code ec;
+
+    try {
+        auto result = co_await ctx.sasl()->authenticate(
+          std::move(request.data.auth_bytes));
+        if (likely(result)) {
+            sasl_authenticate_response_data data{
+              .error_code = error_code::none,
+              .error_message = std::nullopt,
+              .auth_bytes = std::move(result.value()),
+            };
+            co_return co_await ctx.respond(
+              sasl_authenticate_response(std::move(data)));
+        }
+
+        ec = result.error();
+    } catch (security::scram_exception& e) {
+        vlog(
+          klog.warn,
+          "[{}:{}]  Error processing SASL authentication request for {}: {}",
+          ctx.connection()->client_host(),
+          ctx.connection()->client_port(),
+          ctx.header().client_id.value_or(std::string_view("unset-client-id")),
+          e);
+
+        ec = make_error_code(security::errc::invalid_credentials);
     }
 
     sasl_authenticate_response_data data{
       .error_code = error_code::sasl_authentication_failed,
       .error_message = ssx::sformat(
-        "SASL authentication failed: {}", result.error().message()),
+        "SASL authentication failed: {}", ec.message()),
     };
     co_return co_await ctx.respond(sasl_authenticate_response(std::move(data)));
 }