rgw: RGWRadosPutObj evals source bucket perm for backward compatibility

As of a3f40b4 we no longer evaluate perms locally for source bucket,
this could cause broken permission evaluation dusring upgrade as one
zone is not respecting the perm evaluation based on the `rgwx-perm-check-uid`
arg.

This can be dropped in T+2 release.

Signed-off-by: Seena Fallah <[email protected]>

Author: clwluvw

src/rgw/driver/rados/rgw_rados.cc

@@ -3509,7 +3509,6 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
 {
   const DoutPrefixProvider *dpp;
   CephContext* cct;
-  rgw_obj obj;
   rgw::sal::DataProcessor *filter;
   boost::optional<RGWPutObj_Compress>& compressor;
   bool try_etag_verify;
@@ -3528,6 +3527,18 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
   uint64_t ofs{0};
   uint64_t lofs{0}; /* logical ofs */
   std::function<int(map<string, bufferlist>&)> attrs_handler;
+  /*
+    src_bucket_perms, if provided, serves as a fallback mechanism to validate the
+    source bucket's ACL before initiating the fetch operation. It ensures
+    backward compatibility with legacy implementations where the `rgwx-perm-check-uid`
+    arg might not be honored, and the destination zone does not return the
+    `x-rgw-perm-checked` header, which confirms that the source object's
+    permissions were validated against the specified `rgwx-perm-check-uid`.
+
+    @todo drop src_bucket_perms and src_obj in T+2 release.
+  */
+ const rgw_obj& src_obj;
+ const std::optional<RGWUserPermHandler::Bucket>& src_bucket_perms;
 
 public:
   RGWRadosPutObj(const DoutPrefixProvider *dpp,
@@ -3537,7 +3548,9 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
                  rgw::sal::ObjectProcessor *p,
                  void (*_progress_cb)(off_t, void *),
                  void *_progress_data,
-                 std::function<int(map<string, bufferlist>&)> _attrs_handler) :
+                 std::function<int(map<string, bufferlist>&)> _attrs_handler,
+                 const rgw_obj& src_obj,
+                 const std::optional<RGWUserPermHandler::Bucket>& _src_bucket_perms = std::nullopt) :
                        dpp(dpp),
                        cct(cct),
                        filter(p),
@@ -3547,7 +3560,9 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
                        processor(p),
                        progress_cb(_progress_cb),
                        progress_data(_progress_data),
-                       attrs_handler(_attrs_handler) {}
+                       attrs_handler(_attrs_handler),
+                       src_obj(src_obj),
+                       src_bucket_perms(_src_bucket_perms) {}
 
 
   int process_attrs(void) {
@@ -3730,6 +3745,23 @@ class RGWRadosPutObj : public RGWHTTPStreamRWRequest::ReceiveCB
       return "";
     }
   }
+
+  int handle_headers(const map<string, string>& headers, int http_status) override {
+    if (src_bucket_perms && http_status != 403 && http_status != 401) {
+      auto iter = headers.find("RGWX_PERM_CHECKED");
+      // if the header is not present, we need to check the ACL
+      // of the source bucket against the fallback UID
+      if (iter == headers.end()) {
+        // old ACL should still win s3:ReplicateObject
+        if (!src_bucket_perms->verify_bucket_permission(src_obj.key, rgw::IAM::s3ReplicateObject)) {
+          ldpp_dout(dpp, 4) << "ERROR: " << __func__ << "(): fallback permission check denied" << dendl;
+          return -EACCES;
+        }
+      }
+    }
+
+    return 0;
+  }
 };
 
 /*
@@ -4381,6 +4413,32 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& dest_obj_ctx,
 
   std::optional<rgw_user> override_owner;
 
+  // load src bucket perm handler for backward compatibility
+  // @todo drop me in t+2 release.
+  std::optional<RGWUserPermHandler::Bucket> src_bucket_perms;
+  if (perm_check_uid) {
+    std::unique_ptr<rgw::sal::Bucket> src_bucket;
+    ret = driver->load_bucket(rctx.dpp, src_obj.bucket, &src_bucket, rctx.y);
+    if (ret < 0) {
+      ldpp_dout(rctx.dpp, 0) << "ERROR: failed to load src bucket=" << src_obj << " ret=" << ret << dendl;
+      return ret;
+    }
+
+    RGWUserPermHandler user_perms(rctx.dpp, driver, cct, *perm_check_uid);
+    ret = user_perms.init();
+    if (ret < 0) {
+      ldpp_dout(rctx.dpp, 0) << "ERROR: failed to init user perms for: " << src_obj << " ret=" << ret << dendl;
+      return ret;
+    }
+
+    src_bucket_perms.emplace();
+    ret = user_perms.init_bucket(src_bucket->get_info(), src_bucket->get_attrs(), &*src_bucket_perms);
+    if (ret < 0) {
+      ldpp_dout(rctx.dpp, 0) << "ERROR: failed to init bucket perms for: " << src_obj << " ret=" << ret << dendl;
+      return ret;
+    }
+  }
+
   RGWRadosPutObj cb(rctx.dpp, cct, plugin, compressor, &processor, progress_cb, progress_data,
                     [&](map<string, bufferlist>& obj_attrs) {
                       const rgw_placement_rule *ptail_rule;
@@ -4414,7 +4472,7 @@ int RGWRados::fetch_remote_obj(RGWObjectCtx& dest_obj_ctx,
                         return ret;
                       }
                       return 0;
-                    });
+                    }, src_obj, src_bucket_perms);
 
   string etag;
   real_time set_mtime;
@@ -5349,7 +5407,7 @@ int RGWRados::restore_obj_from_cloud(RGWLCCloudTierCtx& tier_ctx,
                       }
                       cb_processed = true;
                       return 0;
-                    });
+                    }, dest_obj);
 
   // fetch mtime of the object and other attrs of the object
   // to check for restore_status

src/test/rgw/rgw_multi/tests.py

@@ -4247,7 +4247,7 @@ def test_bucket_replication_reject_versioning_identical():
     assert e.response['ResponseMetadata']['HTTPStatusCode'] == 400
 
 @allow_bucket_replication
-def test_bucket_replicaion_reject_objectlock_identical():
+def test_bucket_replication_reject_objectlock_identical():
     zonegroup = realm.master_zonegroup()
     zonegroup_conns = ZonegroupConns(zonegroup)