Support specifying users for s3 settings

src/Backups/BackupIO_S3.cpp

@@ -127,7 +127,7 @@ BackupReaderS3::BackupReaderS3(
     : BackupReaderDefault(read_settings_, write_settings_, getLogger("BackupReaderS3"))
     , s3_uri(s3_uri_)
     , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false}
-    , s3_settings(context_->getStorageS3Settings().getSettings(s3_uri.uri.toString()))
+    , s3_settings(context_->getStorageS3Settings().getSettings(s3_uri.uri.toString(), context_->getUserName()))
 {
     auto & request_settings = s3_settings.request_settings;
     request_settings.updateFromSettings(context_->getSettingsRef());
@@ -217,7 +217,7 @@ BackupWriterS3::BackupWriterS3(
     : BackupWriterDefault(read_settings_, write_settings_, getLogger("BackupWriterS3"))
     , s3_uri(s3_uri_)
     , data_source_description{DataSourceType::ObjectStorage, ObjectStorageType::S3, MetadataStorageType::None, s3_uri.endpoint, false, false}
-    , s3_settings(context_->getStorageS3Settings().getSettings(s3_uri.uri.toString()))
+    , s3_settings(context_->getStorageS3Settings().getSettings(s3_uri.uri.toString(), context_->getUserName()))
 {
     auto & request_settings = s3_settings.request_settings;
     request_settings.updateFromSettings(context_->getSettingsRef());

src/IO/S3Common.cpp

@@ -124,6 +124,15 @@ AuthSettings AuthSettings::loadFromConfig(const std::string & config_elem, const
     HTTPHeaderEntries headers = getHTTPHeaders(config_elem, config);
     ServerSideEncryptionKMSConfig sse_kms_config = getSSEKMSConfig(config_elem, config);
 
+    std::unordered_set<std::string> users;
+    Poco::Util::AbstractConfiguration::Keys keys;
+    config.keys(config_elem, keys);
+    for (const auto & key : keys)
+    {
+        if (startsWith(key, "user"))
+            users.insert(config.getString(config_elem + "." + key));
+    }
+
     return AuthSettings
     {
         std::move(access_key_id), std::move(secret_access_key), std::move(session_token),
@@ -134,10 +143,16 @@ AuthSettings AuthSettings::loadFromConfig(const std::string & config_elem, const
         use_environment_credentials,
         use_insecure_imds_request,
         expiration_window_seconds,
-        no_sign_request
+        no_sign_request,
+        std::move(users)
     };
 }
 
+bool AuthSettings::canBeUsedByUser(const String & user) const
+{
+    return users.empty() || users.contains(user);
+}
+
 bool AuthSettings::hasUpdates(const AuthSettings & other) const
 {
     AuthSettings copy = *this;
@@ -173,6 +188,8 @@ void AuthSettings::updateFrom(const AuthSettings & from)
 
     if (from.no_sign_request.has_value())
         no_sign_request = from.no_sign_request;
+
+    users.insert(from.users.begin(), from.users.end());
 }
 
 }

src/IO/S3Common.h

@@ -92,9 +92,13 @@ struct AuthSettings
     std::optional<uint64_t> expiration_window_seconds;
     std::optional<bool> no_sign_request;
 
+    std::unordered_set<std::string> users;
+
     bool hasUpdates(const AuthSettings & other) const;
     void updateFrom(const AuthSettings & from);
 
+    bool canBeUsedByUser(const String & user) const;
+
 private:
     bool operator==(const AuthSettings & other) const = default;
 };

src/Storages/StorageS3.cpp

@@ -1385,7 +1385,7 @@ const StorageS3::Configuration & StorageS3::getConfiguration()
 
 bool StorageS3::Configuration::update(const ContextPtr & context)
 {
-    auto s3_settings = context->getStorageS3Settings().getSettings(url.uri.toString());
+    auto s3_settings = context->getStorageS3Settings().getSettings(url.uri.toString(), context->getUserName());
     request_settings = s3_settings.request_settings;
     request_settings.updateFromSettings(context->getSettings());
 

src/Storages/StorageS3Settings.cpp

@@ -293,7 +293,7 @@ void StorageS3Settings::loadFromConfig(const String & config_elem, const Poco::U
     }
 }
 
-S3Settings StorageS3Settings::getSettings(const String & endpoint) const
+S3Settings StorageS3Settings::getSettings(const String & endpoint, const String & user) const
 {
     std::lock_guard lock(mutex);
     auto next_prefix_setting = s3_settings.upper_bound(endpoint);
@@ -302,7 +302,8 @@ S3Settings StorageS3Settings::getSettings(const String & endpoint) const
     for (auto possible_prefix_setting = next_prefix_setting; possible_prefix_setting != s3_settings.begin();)
     {
         std::advance(possible_prefix_setting, -1);
-        if (boost::algorithm::starts_with(endpoint, possible_prefix_setting->first))
+        const auto & [endpoint_prefix, settings] = *possible_prefix_setting;
+        if (boost::algorithm::starts_with(endpoint, endpoint_prefix) && settings.auth_settings.canBeUsedByUser(user))
             return possible_prefix_setting->second;
     }
 

src/Storages/StorageS3Settings.h

@@ -112,7 +112,7 @@ class StorageS3Settings
 public:
     void loadFromConfig(const String & config_elem, const Poco::Util::AbstractConfiguration & config, const Settings & settings);
 
-    S3Settings getSettings(const String & endpoint) const;
+    S3Settings getSettings(const String & endpoint, const String & user) const;
 
 private:
     mutable std::mutex mutex;

tests/integration/test_backup_restore_s3/configs/s3_settings.xml

@@ -10,6 +10,13 @@
             <upload_part_size_multiply_parts_count_threshold>3</upload_part_size_multiply_parts_count_threshold>
             <upload_part_size_multiply_factor>2</upload_part_size_multiply_factor>
         </multipart>
+        <limited>
+            <endpoint>http://minio1:9001/root/data/backups/limited/</endpoint>
+            <access_key_id>minio</access_key_id>
+            <secret_access_key>minio123</secret_access_key>
+            <user>superuser1</user>
+            <user>superuser2</user>
+        </limited>
     </s3>
     <backup_threads>1</backup_threads>
     <restore_threads>1</restore_threads>

tests/integration/test_backup_restore_s3/test.py

@@ -452,3 +452,57 @@ def test_backup_to_zip():
     backup_name = new_backup_name()
     backup_destination = f"S3('http://minio1:9001/root/data/backups/{backup_name}.zip', 'minio', 'minio123')"
     check_backup_and_restore(storage_policy, backup_destination)
+
+
+def test_user_specific_auth(start_cluster):
+    def create_user(user):
+        node.query(f"CREATE USER {user}")
+        node.query(f"GRANT CURRENT GRANTS ON *.* TO {user}")
+
+    create_user("superuser1")
+    create_user("superuser2")
+    create_user("regularuser")
+
+    node.query("CREATE TABLE specific_auth (col UInt64) ENGINE=Memory")
+
+    assert "Access Denied" in node.query_and_get_error(
+        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')"
+    )
+    assert "Access Denied" in node.query_and_get_error(
+        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        user="regularuser",
+    )
+
+    node.query(
+        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        user="superuser1",
+    )
+    node.query(
+        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        user="superuser1",
+    )
+
+    node.query(
+        "BACKUP TABLE specific_auth TO S3('http://minio1:9001/root/data/backups/limited/backup2.zip')",
+        user="superuser2",
+    )
+    node.query(
+        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup2.zip')",
+        user="superuser2",
+    )
+
+    assert "Access Denied" in node.query_and_get_error(
+        "RESTORE TABLE specific_auth FROM S3('http://minio1:9001/root/data/backups/limited/backup1.zip')",
+        user="regularuser",
+    )
+
+    assert "HTTP response code: 403" in node.query_and_get_error(
+        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1.zip', 'RawBLOB')",
+        user="regularuser",
+    )
+    node.query(
+        "SELECT * FROM s3('http://minio1:9001/root/data/backups/limited/backup1.zip', 'RawBLOB')",
+        user="superuser1",
+    )
+
+    node.query("DROP TABLE IF EXISTS test.specific_auth")