PBS-8: Introduce <replication_source.authentication> section into the main configuration file

CMakeLists.txt

@@ -530,6 +530,10 @@ set(binsrv_source_files
   src/binsrv/basic_logger.hpp
   src/binsrv/basic_logger.cpp
 
+  src/binsrv/authentication_config_fwd.hpp
+  src/binsrv/authentication_config.hpp
+  src/binsrv/authentication_config.cpp
+
   src/binsrv/basic_keyring_fwd.hpp
   src/binsrv/basic_keyring.hpp
   src/binsrv/basic_keyring.cpp

README.md

@@ -517,7 +517,12 @@ The Percona Binary Log Server configuration file has the following format.
   "replication_source": {
     "port": 3307,
     "read_timeout": 60,
-    "write_timeout": 60
+    "write_timeout": 60,
+    "authentication": {
+      "user": "rpl_user",
+      "password": "rpl_password",
+      "plugin": "caching_sha2_password"
+    }
   },
   "keyring": {
     "uri": "file:///var/lib/pbs/keyring/keyring_data.json"
@@ -595,6 +600,12 @@ This section configures the built-in MySQL-compatible listener the utility expos
 - `<replication_source.read_timeout>` - the number of seconds the utility will wait to read data from a connected replica before treating the connection as timed out.
 - `<replication_source.write_timeout>` - the number of seconds the utility will wait to write data to a connected replica before treating the connection as timed out.
 
+#### \<replication_source.authentication\> section
+Credentials the built-in listener accepts from downstream replicas.
+- `<replication_source.authentication.user>` - the MySQL account name a downstream replica must present to log in to the listener (must not be empty).
+- `<replication_source.authentication.password>` - the password associated with that account (must not be empty).
+- `<replication_source.authentication.plugin>` - the client authentication plugin the listener advertises. Only `caching_sha2_password` is supported today; other values are rejected at configuration load time.
+
 #### \<keyring\> section
 If this an optional section that specifies keyring configuration parameters. It must be present if the storage has at least one encrypted binlog file.
 - `<keyring.uri>` - specifies location of the keyring JSON data file (currently only 'file://' scheme is supported meaning that the file should be taken from the local file sytem from the path specified in this URI, e.g. `file:///var/lib/pbs/keyring/keyring_data.json`).

main_config.json

@@ -39,7 +39,12 @@
   "replication_source": {
     "port": 3307,
     "read_timeout": 60,
-    "write_timeout": 60
+    "write_timeout": 60,
+    "authentication": {
+      "user": "rpl_user",
+      "password": "rpl_password",
+      "plugin": "caching_sha2_password"
+    }
   },
   "keyring": {
     "uri": "file:///home/user/keyring/keyring/keyring_data.json"

mtr/binlog_streaming/include/generate_binsrv_config.inc

@@ -27,6 +27,8 @@
 #   --let $binsrv_encryption_cipher = AES-256-CTR (optional)
 #   --let $binsrv_encryption_kek_id = alpha (optional)
 #   --let $binsrv_keyring_data_file_path = $MYSQL_TMP_DIR/keyring_data.json (optional)
+#   --let $binsrv_auth_user = rpl (optional, default: rpl)
+#   --let $binsrv_auth_password = password (optional, default: password)
 #   --source set_up_binsrv_environment.inc
 
 --echo
@@ -107,6 +109,15 @@ if ($binsrv_connection_host == "")
 # connecting a probe client to the listener).
 --let $binsrv_replication_source_port = `SELECT @@global.port + 1000`
 
+if ($binsrv_auth_user == "")
+{
+  --let $binsrv_auth_user = rpl
+}
+if ($binsrv_auth_password == "")
+{
+  --let $binsrv_auth_password = password
+}
+
 eval SET @binsrv_config_json = JSON_OBJECT(
   'logger', JSON_OBJECT(
     'level', '$binsrv_log_level',
@@ -130,7 +141,12 @@ eval SET @binsrv_config_json = JSON_OBJECT(
   'replication_source', JSON_OBJECT(
      'port', $binsrv_replication_source_port,
      'read_timeout', 60,
-     'write_timeout', 60
+     'write_timeout', 60,
+     'authentication', JSON_OBJECT(
+        'user', '$binsrv_auth_user',
+        'password', '$binsrv_auth_password',
+        'plugin', 'caching_sha2_password'
+     )
   ),
   'storage', JSON_OBJECT(
      'backend', '$storage_backend',

src/binsrv/authentication_config.cpp

@@ -0,0 +1,48 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#include "binsrv/authentication_config.hpp"
+
+#include <stdexcept>
+#include <string_view>
+
+#include "util/exception_location_helpers.hpp"
+
+namespace binsrv {
+
+void authentication_config::validate() const {
+  // The only client authentication plugin the PBS listener supports
+  // today. Anything else is rejected up front so that a misconfigured
+  // JSON cannot silently downgrade a session's auth negotiation.
+  static constexpr std::string_view supported_plugin{"caching_sha2_password"};
+
+  if (get<"user">().empty()) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source authentication config: "
+        "user must not be empty");
+  }
+  if (get<"password">().empty()) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source authentication config: "
+        "password must not be empty");
+  }
+  if (get<"plugin">() != supported_plugin) {
+    util::exception_location().raise<std::invalid_argument>(
+        "error validating replication source authentication config: "
+        "plugin must be \"caching_sha2_password\"");
+  }
+}
+
+} // namespace binsrv

src/binsrv/authentication_config.hpp

@@ -0,0 +1,40 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#ifndef BINSRV_AUTHENTICATION_CONFIG_HPP
+#define BINSRV_AUTHENTICATION_CONFIG_HPP
+
+#include "binsrv/authentication_config_fwd.hpp" // IWYU pragma: export
+
+#include <string>
+
+#include "util/nv_tuple.hpp"
+
+namespace binsrv {
+
+struct [[nodiscard]] authentication_config
+    : util::nv_tuple<
+          // clang-format off
+          util::nv<"user"    , std::string>,
+          util::nv<"password", std::string>,
+          util::nv<"plugin"  , std::string>
+          // clang-format on
+          > {
+  void validate() const;
+};
+
+} // namespace binsrv
+
+#endif // BINSRV_AUTHENTICATION_CONFIG_HPP

src/binsrv/authentication_config_fwd.hpp

@@ -0,0 +1,25 @@
+// Copyright (c) 2023-2026 Percona and/or its affiliates.
+//
+// This program is free software; you can redistribute it and/or modify
+// it under the terms of the GNU General Public License, version 2.0,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License, version 2.0, for more details.
+//
+// You should have received a copy of the GNU General Public License
+// along with this program; if not, write to the Free Software
+// Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+#ifndef BINSRV_AUTHENTICATION_CONFIG_FWD_HPP
+#define BINSRV_AUTHENTICATION_CONFIG_FWD_HPP
+
+namespace binsrv {
+
+struct authentication_config;
+
+} // namespace binsrv
+
+#endif // BINSRV_AUTHENTICATION_CONFIG_FWD_HPP

src/binsrv/replication_source_config.cpp

@@ -37,6 +37,7 @@ void replication_source_config::validate() const {
         "error validating replication source config: "
         "write_timeout must be greater than 0");
   }
+  get<"authentication">().validate();
 }
 
 } // namespace binsrv

src/binsrv/replication_source_config.hpp

@@ -20,16 +20,19 @@
 
 #include <cstdint>
 
+#include "binsrv/authentication_config.hpp" // IWYU pragma: export
+
 #include "util/nv_tuple.hpp"
 
 namespace binsrv {
 
 struct [[nodiscard]] replication_source_config
     : util::nv_tuple<
           // clang-format off
-          util::nv<"port"         , std::uint16_t>,
-          util::nv<"read_timeout" , std::uint32_t>,
-          util::nv<"write_timeout", std::uint32_t>
+          util::nv<"port"          , std::uint16_t>,
+          util::nv<"read_timeout"  , std::uint32_t>,
+          util::nv<"write_timeout" , std::uint32_t>,
+          util::nv<"authentication", authentication_config>
           // clang-format on
           > {
   void validate() const;

src/minimysql/network_service.cpp

@@ -60,8 +60,10 @@
 
 #include <boost/system/system_error.hpp>
 
+#include "binsrv/authentication_config.hpp"
 #include "binsrv/basic_logger.hpp"
 #include "binsrv/log_severity.hpp"
+#include "binsrv/replication_source_config.hpp"
 #include "binsrv/storage.hpp"
 
 #include "minimysql/connection_context.hpp"
@@ -281,24 +283,25 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) {
     binsrv::basic_logger &logger,
     // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
     binsrv::storage &storage, boost::asio::ip::tcp::socket socket,
-    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
-    std::chrono::seconds read_timeout, std::chrono::seconds write_timeout,
     // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
-    const std::string &username,
-    // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
-    const std::string &password) {
+    const binsrv::replication_source_config &cfg) {
   boost::system::error_code session_ec;
   const auto remote_endpoint{socket.remote_endpoint(session_ec)};
   const auto remote_endpoint_str{
       boost::lexical_cast<std::string>(remote_endpoint)};
 
   const scope_tracer tracer(logger, "session " + remote_endpoint_str);
 
+  const std::chrono::seconds read_timeout{cfg.get<"read_timeout">()};
+  const std::chrono::seconds write_timeout{cfg.get<"write_timeout">()};
+
   try {
     minimysql::network_buffer_type data;
     data.reserve(network_service::expected_packet_size);
 
-    minimysql::connection_context context{username, password};
+    minimysql::connection_context context{
+        cfg.get<"authentication">().get<"user">(),
+        cfg.get<"authentication">().get<"password">()};
 
     // creating and sending server greeting packet:
     //   protocol_version: 10
@@ -566,12 +569,8 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) {
     binsrv::storage &storage,
     // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
     boost::asio::ip::tcp::acceptor &acceptor,
-    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
-    std::chrono::seconds read_timeout, std::chrono::seconds write_timeout,
-    // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
-    const std::string &username,
     // NOLINTNEXTLINE(cppcoreguidelines-avoid-reference-coroutine-parameters)
-    const std::string &password) {
+    const binsrv::replication_source_config &cfg) {
   const scope_tracer tracer(logger, "listener");
 
   auto executor = acceptor.get_executor();
@@ -597,9 +596,7 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) {
 
       // NOLINTNEXTLINE(misc-include-cleaner)
       boost::asio::co_spawn(executor,
-                            session(logger, storage, std::move(socket),
-                                    read_timeout, write_timeout, username,
-                                    password),
+                            session(logger, storage, std::move(socket), cfg),
                             boost::asio::detached);
     }
   } catch (...) {
@@ -609,23 +606,19 @@ void handle_exception(binsrv::basic_logger &logger, std::string_view context) {
 
 } // anonymous namespace
 
-network_service::network_service(
-    binsrv::basic_logger_ptr logger, boost::asio::io_context &context,
-    binsrv::storage_ptr storage, std::uint16_t listening_port,
-    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
-    std::chrono::seconds read_timeout, std::chrono::seconds write_timeout,
-    // NOLINTNEXTLINE(bugprone-easily-swappable-parameters)
-    std::string_view username, std::string_view password)
+network_service::network_service(binsrv::basic_logger_ptr logger,
+                                 boost::asio::io_context &context,
+                                 binsrv::storage_ptr storage,
+                                 const binsrv::replication_source_config &cfg)
     : logger_{std::move(logger)}, storage_{std::move(storage)},
-      username_(username), password_(password), context_{&context},
+      context_{&context},
       acceptor_{std::make_unique<acceptor_type>(
           context, boost::asio::ip::tcp::endpoint{boost::asio::ip::tcp::v4(),
-                                                  listening_port})} {
+                                                  cfg.get<"port">()})} {
   assert(logger_);
   // NOLINTNEXTLINE(misc-include-cleaner)
   boost::asio::co_spawn(*context_,
-                        listener(*logger_, *storage_, *acceptor_, read_timeout,
-                                 write_timeout, username_, password_),
+                        listener(*logger_, *storage_, *acceptor_, cfg),
                         boost::asio::detached);
 }
 

src/minimysql/network_service.hpp

@@ -16,14 +16,10 @@
 #ifndef MINIMYSQL_NETWORK_SERVICE_HPP
 #define MINIMYSQL_NETWORK_SERVICE_HPP
 
-#include <chrono>
-#include <cstdint>
-#include <string>
-#include <string_view>
-
 #include <boost/asio/ts/netfwd.hpp>
 
 #include "binsrv/basic_logger_fwd.hpp"
+#include "binsrv/replication_source_config_fwd.hpp"
 #include "binsrv/storage_fwd.hpp"
 
 namespace minimysql {
@@ -34,10 +30,7 @@ class network_service {
 
   network_service(binsrv::basic_logger_ptr logger,
                   boost::asio::io_context &context, binsrv::storage_ptr storage,
-                  std::uint16_t listening_port,
-                  std::chrono::seconds read_timeout,
-                  std::chrono::seconds write_timeout, std::string_view username,
-                  std::string_view password);
+                  const binsrv::replication_source_config &cfg);
 
   network_service(const network_service &) = delete;
   network_service &operator=(const network_service &) = delete;
@@ -49,8 +42,6 @@ class network_service {
 private:
   binsrv::basic_logger_ptr logger_;
   binsrv::storage_ptr storage_;
-  std::string username_;
-  std::string password_;
 
   boost::asio::io_context *context_;
   using acceptor_type =