Mask password in data_path in the system.distribution_queue

programs/server/config.d/test_cluster_with_incorrect_pw.xml

@@ -0,0 +1 @@
+../../../tests/config/config.d/test_cluster_with_incorrect_pw.xml

src/Storages/System/StorageSystemDistributionQueue.cpp

@@ -10,6 +10,75 @@
 #include <Common/typeid_cast.h>
 #include <Databases/IDatabase.h>
 
+namespace DB
+{
+
+namespace ErrorCodes
+{
+    extern const int LOGICAL_ERROR;
+}
+
+}
+
+
+namespace
+{
+
+using namespace DB;
+
+/// Drop "password" from the path.
+///
+/// In case of use_compact_format_in_distributed_parts_names=0 the path format is:
+///
+///     user[:password]@host:port#default_database format
+///
+/// And password should be masked out.
+///
+/// See:
+/// - Cluster::Address::fromFullString()
+/// - Cluster::Address::toFullString()
+std::string maskDataPath(const std::string & path)
+{
+    std::string masked_path = path;
+
+    if (!masked_path.ends_with('/'))
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Invalid path format");
+
+    masked_path.pop_back();
+
+    size_t dir_name_pos = masked_path.rfind('/');
+    if (dir_name_pos == std::string::npos)
+    {
+        /// Do not include full path into the exception message since it may include password.
+        throw Exception(ErrorCodes::LOGICAL_ERROR, "Invalid path format");
+    }
+    ++dir_name_pos;
+
+    size_t user_pw_end = masked_path.find('@', dir_name_pos);
+    if (user_pw_end == std::string::npos)
+    {
+        /// Likey new format (use_compact_format_in_distributed_parts_names=1)
+        return path;
+    }
+
+    size_t pw_start = masked_path.find(':', dir_name_pos);
+    if (pw_start > user_pw_end)
+    {
+        /// No password in path
+        return path;
+    }
+    ++pw_start;
+
+    size_t pw_length = user_pw_end - pw_start;
+    /// Replace with a single '*' to hide even the password length.
+    masked_path.replace(pw_start, pw_length, 1, '*');
+
+    masked_path.push_back('/');
+
+    return masked_path;
+}
+
+}
 
 namespace DB
 {
@@ -103,7 +172,7 @@ void StorageSystemDistributionQueue::fillData(MutableColumns & res_columns, cons
             size_t col_num = 0;
             res_columns[col_num++]->insert(database);
             res_columns[col_num++]->insert(table);
-            res_columns[col_num++]->insert(status.path);
+            res_columns[col_num++]->insert(maskDataPath(status.path));
             res_columns[col_num++]->insert(status.is_blocked);
             res_columns[col_num++]->insert(status.error_count);
             res_columns[col_num++]->insert(status.files_count);

tests/config/config.d/test_cluster_with_incorrect_pw.xml

@@ -0,0 +1,14 @@
+<yandex>
+    <remote_servers>
+        <test_cluster_with_incorrect_pw>
+             <shard>
+                 <replica>
+                     <host>localhost</host>
+                     <port>9000</port>
+                     <!-- password is incorrect -->
+                     <password>foo</password>
+                 </replica>
+             </shard>
+         </test_cluster_with_incorrect_pw>
+    </remote_servers>
+</yandex>

tests/config/install.sh

@@ -27,6 +27,7 @@ ln -sf $SRC_PATH/config.d/secure_ports.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/clusters.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/graphite.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/database_atomic.xml $DEST_SERVER_PATH/config.d/
+ln -sf $SRC_PATH/config.d/test_cluster_with_incorrect_pw.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/users.d/log_queries.xml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/readonly.xml $DEST_SERVER_PATH/users.d/
 ln -sf $SRC_PATH/users.d/access_management.xml $DEST_SERVER_PATH/users.d/

tests/queries/0_stateless/01555_system_distribution_queue_mask.reference

@@ -0,0 +1,4 @@
+masked
+2,"default:*@localhost:9000"
+no masking
+1,"default@localhost:9000"

tests/queries/0_stateless/01555_system_distribution_queue_mask.sql

@@ -0,0 +1,36 @@
+-- force data path with the user/pass in it
+set use_compact_format_in_distributed_parts_names=0;
+-- use async send even for localhost
+set prefer_localhost_replica=0;
+
+drop table if exists dist_01555;
+drop table if exists data_01555;
+create table data_01555 (key Int) Engine=Null();
+
+--
+-- masked
+--
+SELECT 'masked';
+create table dist_01555 (key Int) Engine=Distributed(test_cluster_with_incorrect_pw, currentDatabase(), data_01555, key);
+
+insert into dist_01555 values (1)(2);
+-- since test_cluster_with_incorrect_pw contains incorrect password ignore error
+system flush distributed dist_01555; -- { serverError 516; }
+select length(splitByChar('*', data_path)), replaceRegexpOne(data_path, '^.*/([^/]*)/' , '\\1') from system.distribution_queue where database = currentDatabase() and table = 'dist_01555' format CSV;
+
+drop table dist_01555;
+
+--
+-- no masking
+--
+SELECT 'no masking';
+create table dist_01555 (key Int) Engine=Distributed(test_shard_localhost, currentDatabase(), data_01555, key);
+
+insert into dist_01555 values (1)(2);
+-- since test_cluster_with_incorrect_pw contains incorrect password ignore error
+system flush distributed dist_01555;
+select length(splitByChar('*', data_path)), replaceRegexpOne(data_path, '^.*/([^/]*)/' , '\\1') from system.distribution_queue where database = currentDatabase() and table = 'dist_01555' format CSV;
+
+-- cleanup
+drop table dist_01555;
+drop table data_01555;

tests/queries/0_stateless/arcadia_skip_list.txt

@@ -159,4 +159,5 @@
 01547_query_log_current_database
 01548_query_log_query_execution_ms
 01552_dict_fixedstring
+01555_system_distribution_queue_mask
 01557_max_parallel_replicas_no_sample.sql