Merge pull request #5007 from andrwng/sasl-scram-v21.11.x

[v21.11.x] security: don't log potentially sensitive messages

src/v/kafka/protocol/batch_reader.h

@@ -54,6 +54,7 @@ class batch_reader final : public model::record_batch_reader::impl {
     ss::future<storage_t> do_load_slice(model::timeout_clock::time_point) final;
 
     // Implements model::record_batch_reader::impl
+    // NOTE: this stream is intentially devoid of user data.
     void print(std::ostream& os) final { os << "{kafka::consumer_records}"; }
 
     // Release any remaining iobuf that hasn't been consumed
@@ -65,6 +66,7 @@ class batch_reader final : public model::record_batch_reader::impl {
 };
 
 inline std::ostream& operator<<(std::ostream& os, const batch_reader& reader) {
+    // NOTE: this stream is intentially devoid of user data.
     fmt::print(os, "{{size {}}}", reader.size_bytes());
     return os;
 }

src/v/kafka/protocol/kafka_batch_adapter.h

@@ -99,6 +99,7 @@ struct produce_request_record_data {
 
 inline std::ostream&
 operator<<(std::ostream& os, const produce_request_record_data& data) {
+    // NOTE: this stream is intentially devoid of user data.
     fmt::print(
       os,
       "batch {} v2_format {} valid_crc {}",

src/v/kafka/protocol/schemata/generator.py

@@ -32,6 +32,12 @@
 #   ignorable flag on a field doesn't change the wire protocol, but gives
 #   instruction on how things should behave when there is missing data.
 #
+#   - Build a more robust way to define messages and their contents as
+#   sensitive. The current approach relies on a list of root structs to
+#   generate stream operators that don't contain sensitive information. We
+#   should strongly consider leveraging the C++ type system to generate code
+#   that forces users to explicitly "unwrap" sensitive fields before use, to
+#   make it easier to audit where sensitive information is used.
 import io
 import json
 import functools
@@ -290,6 +296,18 @@
     fetch_record_set=("batch_reader", None, "read_nullable_batch_reader()"),
 )
 
+# Declare some fields as sensitive. Utmost care should be taken to ensure the
+# contents of these fields are never made available over unsecure channels
+# (sent over an unencrypted connection, printed in logs, etc.).
+sensitive_map = {
+    "SaslAuthenticateRequestData": {
+        "AuthBytes": True,
+    },
+    "SaslAuthenticateResponseData": {
+        "AuthBytes": True,
+    },
+}
+
 # apply a rename to a struct. this is useful when there is a type name conflict
 # between two request types. since we generate types in a flat namespace this
 # feature is important for resolving naming conflicts.
@@ -683,6 +701,23 @@ def decoder(self):
         assert plain_decoder[1]
         return plain_decoder[1], named_type
 
+    @property
+    def is_sensitive(self):
+        d = sensitive_map
+        for p in self._path:
+            d = d.get(p, None)
+            if d is None:
+                break
+        if type(d) is dict:
+            # We got to the end of this field's path and `sensitive_map` has
+            # sensitive decendents defined. This field is an ancestor of a
+            # sensitive field, but it itself isn't sensitive.
+            return False
+        assert d is None or d is True, \
+          "expected field '{}' to be missing or True; field path: {}, remaining path: {}" \
+          .format(self._field["name"], self._path, d)
+        return d
+
     @property
     def is_array(self):
         return isinstance(self._type, ArrayType)
@@ -909,11 +944,12 @@ class response;
 {% set structs = struct.structs() + [struct] %}
 {% for struct in structs %}
 {%- if struct.fields %}
-std::ostream& operator<<(std::ostream& o, const {{ struct.name }}& v) {
+std::ostream& operator<<(std::ostream& o, [[maybe_unused]] const {{ struct.name }}& v) {
     fmt::print(o,
       "{{'{{' + struct.format + '}}'}}",
       {%- for field in struct.fields %}
-      v.{{ field.name }}{% if not loop.last %},{% endif %}
+      {%- if field.is_sensitive %}"****"{% else %}v.{{ field.name }}{% endif %}{% if not loop.last %},{% endif %}
+
       {%- endfor %}
     );
     return o;

src/v/kafka/protocol/tests/security_test.cc

@@ -8,9 +8,12 @@
 // by the Apache License, Version 2.0
 #include "security/acl.h"
 #define BOOST_TEST_MODULE example
+#include "kafka/protocol/schemata/sasl_authenticate_request.h"
+#include "kafka/protocol/schemata/sasl_authenticate_response.h"
 #include "kafka/server/handlers/details/security.h"
 #include "random/generators.h"
 #include "security/authorizer.h"
+#include "ssx/sformat.h"
 #include "utils/base64.h"
 
 #include <seastar/testing/thread_test_case.hh>
@@ -127,4 +130,15 @@ BOOST_AUTO_TEST_CASE(to_acl_permission) {
       details::to_acl_permission(4), details::acl_conversion_error);
 }
 
+BOOST_AUTO_TEST_CASE(redact_sensitive_messages) {
+    BOOST_REQUIRE_EQUAL(
+      "{auth_bytes=****}", ssx::sformat("{}", sasl_authenticate_request_data{}));
+
+    BOOST_REQUIRE_EQUAL(
+      "{error_code={ error_code: none [0] } error_message={nullopt} "
+      "auth_bytes=**** "
+      "session_lifetime_ms=0}",
+      ssx::sformat("{}", sasl_authenticate_response_data{}));
+}
+
 } // namespace kafka

src/v/kafka/server/handlers/sasl_authenticate.cc

@@ -12,6 +12,7 @@
 #include "bytes/bytes.h"
 #include "kafka/protocol/errors.h"
 #include "security/scram_algorithm.h"
+#include "vlog.h"
 
 namespace kafka {
 

src/v/security/scram_algorithm.cc

@@ -329,17 +329,6 @@ parse_server_final(std::string_view message) {
 
 namespace security {
 
-std::ostream& operator<<(std::ostream& os, const scram_credential& cred) {
-    fmt::print(
-      os,
-      "salt {} server_key {} stored_key {} iterations {}",
-      cred._salt,
-      cred._server_key,
-      cred._stored_key,
-      cred._iterations);
-    return os;
-}
-
 client_first_message::client_first_message(bytes_view data) {
     auto view = std::string_view(
       reinterpret_cast<const char*>(data.data()), data.size()); // NOLINT
@@ -395,29 +384,21 @@ bool client_first_message::token_authenticated() const {
     return false;
 }
 
-std::ostream& operator<<(std::ostream& os, const client_first_message& msg) {
-    fmt::print(
-      os,
-      "authzid {} username {} nonce {}",
-      msg._authzid,
-      msg._username,
-      msg._nonce);
-    return os;
+std::ostream& operator<<(std::ostream& os, const client_first_message&) {
+    // NOTE: this stream is intentially left minimal to err away from exposing
+    // anything that may be useful for an attacker to use.
+    return os << "{client_first_message}";
 }
 
 ss::sstring server_first_message::sasl_message() const {
     return ssx::sformat(
       "r={},s={},i={}", _nonce, bytes_to_base64(_salt), _iterations);
 }
 
-std::ostream& operator<<(std::ostream& os, const server_first_message& msg) {
-    fmt::print(
-      os,
-      "nonce {} salt {} iterations {}",
-      msg._nonce,
-      msg._salt,
-      msg._iterations);
-    return os;
+std::ostream& operator<<(std::ostream& os, const server_first_message&) {
+    // NOTE: this stream is intentially left minimal to err away from exposing
+    // anything that may be useful for an attacker to use.
+    return os << "{server_first_message}";
 }
 
 client_final_message::client_final_message(bytes_view data) {
@@ -441,15 +422,10 @@ ss::sstring client_final_message::msg_no_proof() const {
     return ssx::sformat("c={},r={}", bytes_to_base64(_channel_binding), _nonce);
 }
 
-std::ostream& operator<<(std::ostream& os, const client_final_message& msg) {
-    fmt::print(
-      os,
-      "channel {} nonce {} extensions {} proof {}",
-      msg._channel_binding,
-      msg._nonce,
-      msg._extensions,
-      msg._proof);
-    return os;
+std::ostream& operator<<(std::ostream& os, const client_final_message&) {
+    // NOTE: this stream is intentially left minimal to err away from exposing
+    // anything that may be useful for an attacker to use.
+    return os << "{client_final_message}";
 }
 
 ss::sstring server_final_message::sasl_message() const {
@@ -459,9 +435,10 @@ ss::sstring server_final_message::sasl_message() const {
     return ssx::sformat("v={}", bytes_to_base64(_signature));
 }
 
-std::ostream& operator<<(std::ostream& os, const server_final_message& msg) {
-    fmt::print(os, "error {} signature {}", msg._error, msg._signature);
-    return os;
+std::ostream& operator<<(std::ostream& os, const server_final_message&) {
+    // NOTE: this stream is intentially left minimal to err away from exposing
+    // anything that may be useful for an attacker to use.
+    return os << "{server_final_message}";
 }
 
 server_first_message::server_first_message(bytes_view data) {

src/v/security/scram_authenticator.cc

@@ -42,8 +42,7 @@ scram_authenticator<T>::handle_client_first(bytes_view auth_bytes) {
     if (_credential->iterations() < scram::min_iterations) {
         vlog(
           seclog.info,
-          "Requested iterations {} less than minimum {}",
-          _credential->iterations(),
+          "Requested iterations less than minimum {}",
           scram::min_iterations);
         return errc::invalid_credentials;
     }

src/v/security/scram_credential.cc

@@ -15,6 +15,8 @@
 namespace security {
 
 std::ostream& operator<<(std::ostream& os, const scram_credential&) {
+    // NOTE: this stream is intentially left minimal to err away from exposing
+    // anything that may be useful for an attacker to use.
     return os << "{scram_credential}";
 }