Using HeadBucket for backups is dangerous in multi-tenant S3 buckets

storage/innobase/xtrabackup/src/xbcloud/s3.cc

@@ -431,7 +431,7 @@ bool S3_client::create_bucket(const std::string &name) {
   return true;
 }
 
-bool S3_client::probe_api_version_and_lookup(const std::string &bucket) {
+bool S3_client::probe_api_version_and_lookup(const std::string &bucket, const std::string &backup) {
   for (auto lookup : {LOOKUP_DNS, LOOKUP_PATH}) {
     if (bucket_lookup != LOOKUP_AUTO && bucket_lookup != lookup) {
       continue;
@@ -455,7 +455,7 @@ bool S3_client::probe_api_version_and_lookup(const std::string &bucket) {
       api_version = version;
 
       bool exists;
-      if (bucket_exists(bucket.c_str(), exists)) {
+      if (object_exists(bucket.c_str(), backup.c_str(), exists)) {
         msg_ts("%s: Successfully connected.\n", my_progname);
         return true;
       }
@@ -498,6 +498,29 @@ bool S3_client::bucket_exists(const std::string &name, bool &exists) {
   return false;
 }
 
+bool S3_client::object_exists(const std::string &bucket, const std::string &name, bool &exists) {
+  Http_request req(Http_request::HEAD, protocol, hostname(bucket),
+                   bucketname(bucket) + "/" + name);
+  signer->sign_request(hostname(bucket), bucket, req, time(0));
+
+  Http_response resp;
+  if (!http_client->make_request(req, resp)) {
+    return false;
+  }
+
+  if (resp.ok()) {
+    exists = true;
+    return true;
+  }
+
+  if (resp.http_code() == 404) {
+    exists = false;
+    return true;
+  }
+
+  return false;
+}
+
 bool S3_client::upload_object(const std::string &bucket,
                               const std::string &name,
                               const Http_buffer &contents) {

storage/innobase/xtrabackup/src/xbcloud/s3.h

@@ -227,14 +227,16 @@ class S3_client {
 
   void set_api_version(s3_api_version_t version) { api_version = version; }
 
-  bool probe_api_version_and_lookup(const std::string &bucket);
+  bool probe_api_version_and_lookup(const std::string &bucket, const std::string &backup);
 
   bool delete_object(const std::string &bucket, const std::string &name);
 
   bool create_bucket(const std::string &name);
 
   bool bucket_exists(const std::string &name, bool &exists);
 
+  bool object_exists(const std::string &bucket, const std::string &name, bool &exists);
+
   bool upload_object(const std::string &bucket, const std::string &name,
                      const Http_buffer &contents);
 
@@ -303,8 +305,8 @@ class S3_object_store : public Object_store {
   void set_ec2_instance(std::shared_ptr<S3_ec2_instance> instance) {
     s3_client.set_ec2_instance(instance);
   }
-  bool probe_api_version_and_lookup(const std::string &bucket) {
-    return s3_client.probe_api_version_and_lookup(bucket);
+  bool probe_api_version_and_lookup(const std::string &bucket, const std::string &backup) {
+    return s3_client.probe_api_version_and_lookup(bucket, backup);
   }
   virtual bool create_container(const std::string &name) override {
     return s3_client.create_bucket(name);

storage/innobase/xtrabackup/src/xbcloud/xbcloud-t.cc

@@ -51,7 +51,7 @@ TEST(s3_client, basicDNSv4) {
                 HasHeader("Authorization")),
           Response(200, "", std::vector<char *>{})))
       .WillOnce(Return(true));
-  c.probe_api_version_and_lookup("probe-bucket");
+  c.probe_api_version_and_lookup("probe-bucket", "probe-backup");
 
   EXPECT_CALL(
       http_client,
@@ -102,7 +102,7 @@ TEST(s3_client, basicPATHv4) {
                 HasHeader("Authorization")),
           Response(200, "", std::vector<char *>{})))
       .WillOnce(Return(true));
-  c.probe_api_version_and_lookup("probe-bucket");
+  c.probe_api_version_and_lookup("probe-bucket", "probe-backup");
 
   EXPECT_CALL(
       http_client,
@@ -142,7 +142,7 @@ TEST(s3_client, basicEndpoint) {
                 HasHeader("Authorization")),
           Response(200, "", std::vector<char *>{})))
       .WillOnce(Return(true));
-  c.probe_api_version_and_lookup("probe-bucket");
+  c.probe_api_version_and_lookup("probe-bucket", "probe-backup");
 
   EXPECT_CALL(
       http_client,

storage/innobase/xtrabackup/src/xbcloud/xbcloud.cc

@@ -931,21 +931,11 @@ void put_func(put_thread_ctxt_t &cntx) {
 
 bool xbcloud_put(Object_store *store, const std::string &container,
                  const std::string &backup_name) {
-  bool exists;
   std::atomic<bool> has_errors{false};
   Http_buffer buf_md5 = Http_buffer();
   std::string last_file_prefix = backup_name + "/xtrabackup_tablespaces";
   auto last_file_size = last_file_prefix.size();
   bool file_found = false;
-  if (!store->container_exists(container, exists)) {
-    return false;
-  }
-
-  if (!exists) {
-    if (!store->create_container(container)) {
-      return false;
-    }
-  }
 
   std::vector<std::string> object_list;
   if (!store->list_objects_in_directory(container, backup_name, object_list)) {
@@ -1488,7 +1478,7 @@ int main(int argc, char **argv) {
     container_name = opt_s3_bucket;
 
     if (!reinterpret_cast<S3_object_store *>(object_store.get())
-             ->probe_api_version_and_lookup(container_name)) {
+             ->probe_api_version_and_lookup(container_name, backup_name)) {
       return EXIT_FAILURE;
     }
     if (ec2_instance->get_is_ec2_instance_with_profile()) {
@@ -1525,7 +1515,7 @@ int main(int argc, char **argv) {
     container_name = opt_google_bucket;
 
     if (!reinterpret_cast<S3_object_store *>(object_store.get())
-             ->probe_api_version_and_lookup(container_name)) {
+             ->probe_api_version_and_lookup(container_name, backup_name)) {
       return EXIT_FAILURE;
     }
   } else if (opt_storage == AZURE) {