[v22.3.x] Backport #8087 (net: improve logging of TLS & auth errors)

src/v/kafka/server/requests.cc

@@ -17,6 +17,7 @@
 #include "kafka/server/handlers/sasl_handshake.h"
 #include "kafka/server/request_context.h"
 #include "kafka/types.h"
+#include "net/types.h"
 #include "utils/to_string.h"
 #include "vlog.h"
 
@@ -107,10 +108,10 @@ process_result_stages process_generic(
     return handler->handle(std::move(ctx), g);
 }
 
-class kafka_authentication_exception : public std::runtime_error {
+class kafka_authentication_exception : public net::authentication_exception {
 public:
     explicit kafka_authentication_exception(const std::string& m)
-      : std::runtime_error(m) {}
+      : net::authentication_exception(m) {}
 };
 
 /*

src/v/net/connection.cc

@@ -31,6 +31,9 @@ bool is_reconnect_error(const std::system_error& e) {
         switch (v) {
         case GNUTLS_E_PUSH_ERROR:
         case GNUTLS_E_PULL_ERROR:
+        case GNUTLS_E_UNEXPECTED_PACKET:
+        case GNUTLS_E_UNSUPPORTED_VERSION_PACKET:
+        case GNUTLS_E_NO_CIPHER_SUITES:
         case GNUTLS_E_PREMATURE_TERMINATION:
             return true;
         default:
@@ -89,6 +92,18 @@ std::optional<ss::sstring> is_disconnect_exception(std::exception_ptr e) {
     return std::nullopt;
 }
 
+bool is_auth_error(std::exception_ptr e) {
+    try {
+        rethrow_exception(e);
+    } catch (const authentication_exception& e) {
+        return true;
+    } catch (...) {
+        return false;
+    }
+
+    __builtin_unreachable();
+}
+
 connection::connection(
   boost::intrusive::list<connection>& hook,
   ss::sstring name,

src/v/net/connection.h

@@ -31,6 +31,8 @@ namespace net {
 bool is_reconnect_error(const std::system_error& e);
 std::optional<ss::sstring> is_disconnect_exception(std::exception_ptr);
 
+bool is_auth_error(std::exception_ptr);
+
 class connection : public boost::intrusive::list_base_hook<> {
 public:
     connection(

src/v/net/server.cc

@@ -118,13 +118,26 @@ static inline void print_exceptional_future(
     auto disconnected = is_disconnect_exception(ex);
 
     if (!disconnected) {
-        vlog(
-          rpc::rpclog.error,
-          "{} - Error[{}] remote address: {} - {}",
-          proto->name(),
-          ctx,
-          address,
-          ex);
+        if (is_auth_error(ex)) {
+            vlog(
+              rpc::rpclog.warn,
+              "{} - Authentication Failure[{}] remote address: {} - {}",
+              proto->name(),
+              ctx,
+              address,
+              ex);
+        } else {
+            // Authentication exceptions are logged at WARN, not ERROR, because
+            // they generally point to a misbehaving client rather than a fault
+            // in the server.
+            vlog(
+              rpc::rpclog.error,
+              "{} - Error[{}] remote address: {} - {}",
+              proto->name(),
+              ctx,
+              address,
+              ex);
+        }
     } else {
         vlog(
           rpc::rpclog.info,

src/v/net/types.h

@@ -20,4 +20,16 @@ using public_metrics_disabled
   = seastar::bool_class<struct public_metrics_disabled_tag>;
 using clock_type = seastar::lowres_clock;
 
+/**
+ * Subclass this exception for exceptions related to authentication, so that
+ * the `net` layer's error handling can use appropriate severity when
+ * logging the exceptions (we log client errors as WARN, and other unexpected
+ * server-side exceptions at ERROR).
+ */
+class authentication_exception : public std::runtime_error {
+protected:
+    explicit authentication_exception(const std::string& m)
+      : std::runtime_error(m) {}
+};
+
 } // namespace net

src/v/security/scram_algorithm.h

@@ -11,6 +11,7 @@
 #pragma once
 #include "bytes/bytes.h"
 #include "hashing/secure.h"
+#include "net/types.h"
 #include "random/generators.h"
 #include "security/scram_credential.h"
 #include "ssx/sformat.h"
@@ -36,15 +37,10 @@
  */
 namespace security {
 
-class scram_exception final : public std::exception {
+class scram_exception final : public net::authentication_exception {
 public:
     explicit scram_exception(ss::sstring msg) noexcept
-      : _msg(std::move(msg)) {}
-
-    const char* what() const noexcept final { return _msg.c_str(); }
-
-private:
-    ss::sstring _msg;
+      : net::authentication_exception(std::move(msg)) {}
 };
 
 /**