feat redis: add sentinel_password option and make sure that it work for non-subscribe cases

This PR adds support for password authentication when connecting to Redis Sentinel.

Closes: userver-framework#924

Allows configuring Sentinel password via `sentinel_password` option in secdist config.
Example:
```json
{
    "redis_settings": {
        "taxi-tmp": {
            "password": "redispwd",
            "sentinel_password": "sentpwd",  /* <---- `AUTH sentpwd` will be executed on connection to sentinel` */
            "database_index": 5,
            "secure_connection": false,
            "shards": [
                {
                    "name": "mymaster"
                }
            ],
            "sentinels": [
                {
                    "host": "127.0.0.1",
                    "port": 26379
                }
            ]
        }
    }
}
```

Tests: протестировано CI

Pull Request resolved: userver-framework#925
commit_hash:84ec2914e9b505c1655600cf3ce6ccf2281ae75c

Author: tysion

redis/include/userver/storages/redis/component.hpp

@@ -99,6 +99,7 @@ namespace components {
 ///   "redis_settings": {
 ///     "some_name_of_your_database": {
 ///       "password": "the_password_of_your_database",
+///       "sentinel_password": "the_password_for_sentinels_if_any",
 ///       "sentinels": [
 ///         {"host": "the_host1_of_your_database", "port": 11564}
 ///       ],

redis/src/storages/redis/impl/secdist_redis.hpp

@@ -21,6 +21,7 @@ struct RedisSettings {
     std::vector<std::string> shards;
     std::vector<HostPort> sentinels;
     storages::redis::Password password{std::string()};
+    storages::redis::Password sentinel_password{std::string()};
     storages::redis::ConnectionSecurity secure_connection{storages::redis::ConnectionSecurity::kNone};
     std::size_t database_index{0};
 };

redis/src/storages/redis/impl/sentinel.cpp

@@ -147,6 +147,7 @@ std::shared_ptr<Sentinel> Sentinel::CreateSentinel(
     const testsuite::RedisControl& testsuite_redis_control
 ) {
     const auto& password = settings.password;
+    const auto& sentinel_password = settings.sentinel_password;
 
     const std::vector<std::string>& shards = settings.shards;
     LOG_DEBUG() << "shards.size() = " << shards.size();
@@ -157,13 +158,12 @@ std::shared_ptr<Sentinel> Sentinel::CreateSentinel(
     LOG_DEBUG() << "sentinels.size() = " << settings.sentinels.size();
     for (const auto& sentinel : settings.sentinels) {
         LOG_DEBUG() << "sentinel:  host = " << sentinel.host << "  port = " << sentinel.port;
-        // SENTINEL MASTERS/SLAVES works without auth, sentinel has no AUTH command.
         // CLUSTER SLOTS works after auth only. Masters and slaves used instead of
         // sentinels in cluster mode.
         conns.emplace_back(
             sentinel.host,
             sentinel.port,
-            (key_shard_factory.IsClusterStrategy() ? password : Password("")),
+            (key_shard_factory.IsClusterStrategy() ? password : sentinel_password),
             false,
             settings.secure_connection
         );

redis/src/storages/redis/impl/server_test.cpp

@@ -7,7 +7,14 @@
 #include <storages/redis/impl/command.hpp>
 #include <storages/redis/impl/secdist_redis.hpp>
 #include <storages/redis/impl/sentinel.hpp>
+#include <storages/redis/impl/subscribe_sentinel.hpp>
 #include <storages/redis/impl/thread_pools.hpp>
+#include <storages/redis/subscribe_client_impl.hpp>
+#include <userver/dynamic_config/test_helpers.hpp>
+#include <userver/storages/redis/subscribe_client.hpp>
+#include <userver/storages/redis/subscription_token.hpp>
+
+#include <userver/utest/utest.hpp>
 
 USERVER_NAMESPACE_BEGIN
 
@@ -44,6 +51,52 @@ bool IsConnected(const storages::redis::impl::Redis& redis) {
     return redis.GetState() == storages::redis::RedisState::kConnected;
 }
 
+struct MockSentinelServers {
+    static constexpr size_t kRedisThreadCount = 1;
+    static constexpr std::string_view kRedisName = "redis_name";
+
+    void RegisterSentinelMastersSlaves() {
+        std::vector<MockRedisServer::SlaveInfo> slave_infos;
+        std::string redis_name{kRedisName};
+        for (const auto& slave : slaves) {
+            slave_infos.emplace_back(redis_name, kLocalhost, slave.GetPort());
+        }
+
+        for (auto& sentinel : sentinels) {
+            sentinel.RegisterSentinelMastersHandler({{redis_name, kLocalhost, masters[0].GetPort()}});
+            sentinel.RegisterSentinelSlavesHandler(redis_name, slave_infos);
+        }
+    }
+
+    template <class Function>
+    void ForEachServer(const Function& visitor) {
+        for (auto& server : masters) {
+            visitor(server);
+        }
+        for (auto& server : slaves) {
+            visitor(server);
+        }
+        for (auto& server : sentinels) {
+            visitor(server);
+        }
+    }
+
+    MockRedisServer masters[1] = {
+        MockRedisServer{"master0"},
+    };
+    MockRedisServer slaves[2] = {
+        MockRedisServer{"slave0"},
+        MockRedisServer{"slave1"},
+    };
+    MockRedisServer sentinels[3] = {
+        MockRedisServer{"sentinel0"},
+        MockRedisServer{"sentinel1"},
+        MockRedisServer{"sentinel2"},
+    };
+    std::shared_ptr<storages::redis::impl::ThreadPools> thread_pool =
+        std::make_shared<storages::redis::impl::ThreadPools>(1, kRedisThreadCount);
+};
+
 }  // namespace
 
 TEST(Redis, NoPassword) {
@@ -101,6 +154,124 @@ TEST(Redis, AuthTimeout) {
     PeriodicCheck([&] { return !IsConnected(*redis); });
 }
 
+UTEST(Redis, SentinelAuth) {
+    MockSentinelServers mock;
+    mock.RegisterSentinelMastersSlaves();
+    mock.ForEachServer([](auto& server) { server.RegisterPingHandler(); });
+    auto& [masters, slaves, sentinels, thread_pool] = mock;
+
+    secdist::RedisSettings settings;
+    settings.shards = {std::string{MockSentinelServers::kRedisName}};
+    settings.sentinel_password = storages::redis::Password("pass");
+    settings.sentinels.reserve(std::size(sentinels));
+    for (const auto& sentinel : sentinels) {
+        settings.sentinels.emplace_back(kLocalhost, sentinel.GetPort());
+    }
+
+    std::vector<MockRedisServer::HandlerPtr> auth_handlers;
+    auth_handlers.reserve(std::size(sentinels));
+    for (auto& sentinel : sentinels) {
+        auth_handlers.push_back(sentinel.RegisterStatusReplyHandler("AUTH", "OK"));
+    }
+    std::vector<MockRedisServer::HandlerPtr> no_auth_handlers;
+    no_auth_handlers.reserve(std::size(masters) + std::size(slaves));
+    for (auto& server : masters) {
+        no_auth_handlers.push_back(server.RegisterStatusReplyHandler("AUTH", "FAIL"));
+    }
+    for (auto& server : slaves) {
+        no_auth_handlers.push_back(server.RegisterStatusReplyHandler("AUTH", "FAIL"));
+    }
+
+    auto sentinel_client = storages::redis::impl::Sentinel::CreateSentinel(
+        thread_pool, settings, "test_shard_group_name", dynamic_config::GetDefaultSource(), "test_client_name", {""}
+    );
+    sentinel_client->WaitConnectedDebug(std::empty(slaves));
+
+    for (auto& handler : auth_handlers) {
+        EXPECT_TRUE(handler->WaitForFirstReply(kSmallPeriod));
+    }
+
+    for (auto& handler : no_auth_handlers) {
+        EXPECT_FALSE(handler->WaitForFirstReply(kWaitPeriod));
+    }
+
+    for (const auto& sentinel : sentinels) {
+        EXPECT_TRUE(sentinel.WaitForFirstPingReply(kSmallPeriod));
+    }
+}
+
+// TODO: TAXICOMMON-10834. Looks like AUTH to sentinel is not sent in case of SUBSCRIBE
+UTEST(Redis, DISABLED_SentinelAuthSubscribe) {
+    MockSentinelServers mock;
+    mock.RegisterSentinelMastersSlaves();
+    mock.ForEachServer([](auto& server) { server.RegisterPingHandler(); });
+    auto& [masters, slaves, sentinels, thread_pool] = mock;
+    // Sentinels do NOT receive SUBSCRIBE
+    std::vector<MockRedisServer::HandlerPtr> subscribe_handlers;
+    for (auto& server : masters) {
+        subscribe_handlers.push_back(server.RegisterHandlerWithConstReply("SUBSCRIBE", 1));
+    }
+    for (auto& server : slaves) {
+        subscribe_handlers.push_back(server.RegisterHandlerWithConstReply("SUBSCRIBE", 1));
+    }
+
+    secdist::RedisSettings settings;
+    settings.shards = {std::string{MockSentinelServers::kRedisName}};
+    settings.sentinel_password = storages::redis::Password("pass");
+    settings.sentinels.reserve(std::size(sentinels));
+    for (const auto& sentinel : sentinels) {
+        settings.sentinels.emplace_back(kLocalhost, sentinel.GetPort());
+    }
+
+    std::vector<MockRedisServer::HandlerPtr> auth_handlers;
+    auth_handlers.reserve(std::size(sentinels));
+    for (auto& sentinel : sentinels) {
+        auth_handlers.push_back(sentinel.RegisterStatusReplyHandler("AUTH", "OK"));
+    }
+    std::vector<MockRedisServer::HandlerPtr> no_auth_handlers;
+    no_auth_handlers.reserve(std::size(masters) + std::size(slaves));
+    for (auto& server : masters) {
+        no_auth_handlers.push_back(server.RegisterStatusReplyHandler("AUTH", "FAIL"));
+    }
+    for (auto& server : slaves) {
+        no_auth_handlers.push_back(server.RegisterStatusReplyHandler("AUTH", "FAIL"));
+    }
+
+    storages::redis::CommandControl cc{};
+    testsuite::RedisControl redis_control{};
+    auto subscribe_sentinel = storages::redis::impl::SubscribeSentinel::Create(
+        thread_pool,
+        settings,
+        "test_shard_group_name",
+        dynamic_config::GetDefaultSource(),
+        "test_client_name",
+        {""},
+        cc,
+        redis_control
+    );
+    subscribe_sentinel->WaitConnectedDebug(std::empty(slaves));
+    std::shared_ptr<storages::redis::SubscribeClient> client =
+        std::make_shared<storages::redis::SubscribeClientImpl>(std::move(subscribe_sentinel));
+
+    storages::redis::SubscriptionToken::OnMessageCb callback = [](const std::string& channel,
+                                                                  const std::string& message) {
+        EXPECT_TRUE(false) << "Should not be called. Channel = " << channel << ", message = " << message;
+    };
+    auto subscription = client->Subscribe("channel_name", std::move(callback));
+
+    for (auto& handler : subscribe_handlers) {
+        EXPECT_TRUE(handler->WaitForFirstReply(utest::kMaxTestWaitTime));
+    }
+
+    for (auto& handler : auth_handlers) {
+        EXPECT_TRUE(handler->WaitForFirstReply(kSmallPeriod));
+    }
+
+    for (auto& handler : no_auth_handlers) {
+        EXPECT_FALSE(handler->WaitForFirstReply(kWaitPeriod));
+    }
+}
+
 TEST(Redis, Select) {
     MockRedisServer server{"redis_db"};
     auto ping_handler = server.RegisterPingHandler();

redis/src/storages/redis/impl/subscribe_sentinel.cpp

@@ -84,6 +84,7 @@ std::shared_ptr<SubscribeSentinel> SubscribeSentinel::Create(
     const testsuite::RedisControl& testsuite_redis_control
 ) {
     const auto& password = settings.password;
+    const auto& sentinel_password = settings.password;
 
     const std::vector<std::string>& shards = settings.shards;
     LOG_DEBUG() << "shards.size() = " << shards.size();
@@ -96,11 +97,14 @@ std::shared_ptr<SubscribeSentinel> SubscribeSentinel::Create(
     LOG_DEBUG() << "sentinels.size() = " << settings.sentinels.size();
     for (const auto& sentinel : settings.sentinels) {
         LOG_DEBUG() << "sentinel:  host = " << sentinel.host << "  port = " << sentinel.port;
-        // SENTINEL MASTERS/SLAVES works without auth, sentinel has no AUTH command.
         // CLUSTER SLOTS works after auth only. Masters and slaves used instead of
         // sentinels in cluster mode.
         conns.emplace_back(
-            sentinel.host, sentinel.port, (is_cluster_mode ? password : Password("")), false, settings.secure_connection
+            sentinel.host,
+            sentinel.port,
+            (is_cluster_mode ? password : sentinel_password),
+            false,
+            settings.secure_connection
         );
     }
     LOG_DEBUG() << "redis command_control: " << command_control.ToString();

redis/src/storages/redis/redis_secdist.cpp

@@ -35,6 +35,8 @@ RedisMapSettings::RedisMapSettings(const formats::json::Value& doc) {
 
         USERVER_NAMESPACE::secdist::RedisSettings settings;
         settings.password = storages::redis::Password(GetString(client_settings, "password"));
+        settings.sentinel_password =
+            storages::redis::Password(client_settings["sentinel_password"].As<std::string>(""));
         settings.secure_connection = GetValue<bool>(client_settings, "secure_connection", false)
                                          ? USERVER_NAMESPACE::storages::redis::ConnectionSecurity::kTLS
                                          : USERVER_NAMESPACE::storages::redis::ConnectionSecurity::kNone;