rgw: Check S3 object lock date in multi-object delete

Multi-object delete (via the S3 API) will now check each object's retention date in the same way as single object delet does.

Fixes: http://tracker.ceph.com/issues/47586
Signed-off-by: Mark Houghton <mhoughton@microfocus.com>

diff --git a/src/rgw/rgw_common.cc b/src/rgw/rgw_common.cc
index 5e8da7dbc483a42d19385941e8b05a0f05e7cd20..b657598dba6e3f4d0e8f40b9e101a37ef0b22320 100644 (file)
--- a/src/rgw/rgw_common.cc
+++ b/src/rgw/rgw_common.cc
@@ -1490,6 +1490,41 @@ bool verify_object_permission(const DoutPrefixProvider* dpp, struct req_state *s
                                   op);
 }
 
+
+int verify_object_lock(const DoutPrefixProvider* dpp, const rgw::sal::RGWAttrs& attrs, const bool bypass_perm, const bool bypass_governance_mode) {
+  auto aiter = attrs.find(RGW_ATTR_OBJECT_RETENTION);
+  if (aiter != attrs.end()) {
+    RGWObjectRetention obj_retention;
+    try {
+      decode(obj_retention, aiter->second);
+    } catch (buffer::error& err) {
+      ldpp_dout(dpp, 0) << "ERROR: failed to decode RGWObjectRetention" << dendl;
+      return -EIO;
+    }
+    if (ceph::real_clock::to_time_t(obj_retention.get_retain_until_date()) > ceph_clock_now()) {
+      if (obj_retention.get_mode().compare("GOVERNANCE") != 0 || !bypass_perm || !bypass_governance_mode) {
+        return -EACCES;
+      }
+    }
+  }
+  aiter = attrs.find(RGW_ATTR_OBJECT_LEGAL_HOLD);
+  if (aiter != attrs.end()) {
+    RGWObjectLegalHold obj_legal_hold;
+    try {
+      decode(obj_legal_hold, aiter->second);
+    } catch (buffer::error& err) {
+      ldpp_dout(dpp, 0) << "ERROR: failed to decode RGWObjectLegalHold" << dendl;
+      return -EIO;
+    }
+    if (obj_legal_hold.is_enabled()) {
+      return -EACCES;
+    }
+  }
+  
+  return 0;
+}
+
+
 class HexTable
 {
   char table[256];

diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index 8658a0020ff03d175e7ebdcbc86db8907f9d0776..dd62c3e254b641f3ab55886eff0327ea513d95ba 100644 (file)
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -48,6 +48,7 @@ namespace rgw::sal {
   class RGWUser;
   class RGWBucket;
   class RGWObject;
+  using RGWAttrs = std::map<std::string, ceph::buffer::list>;
 }
 
 using ceph::crypto::MD5;
@@ -2139,6 +2140,12 @@ extern bool verify_object_permission_no_policy(
   int perm);
 extern bool verify_object_permission_no_policy(const DoutPrefixProvider* dpp, struct req_state *s,
                                               int perm);
+extern int verify_object_lock(
+  const DoutPrefixProvider* dpp,
+  const rgw::sal::RGWAttrs& attrs,
+  const bool bypass_perm,
+  const bool bypass_governance_mode);
+
 /** Convert an input URL into a sane object name
  * by converting %-escaped strings into characters, etc*/
 extern void rgw_uri_escape_char(char c, string& dst);

diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index b7765c9deae1e85232756b283e5ec93d729ee5c2..b259f0c36702698331b33b30f67674f26739f99e 100644 (file)
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -4748,13 +4748,13 @@ void RGWDeleteObj::execute(optional_yield y)
       }
 
       if (check_obj_lock) {
-       /* check if obj exists, read orig attrs */
-       if (op_ret == -ENOENT) {
-         /* object maybe delete_marker, skip check_obj_lock*/
-         check_obj_lock = false;
-       } else {
-         return;
-       }
+        /* check if obj exists, read orig attrs */
+        if (op_ret == -ENOENT) {
+          /* object maybe delete_marker, skip check_obj_lock*/
+          check_obj_lock = false;
+        } else {
+          return;
+        }
       }
     } else {
       attrs = s->object->get_attrs();
@@ -4764,37 +4764,10 @@ void RGWDeleteObj::execute(optional_yield y)
     op_ret = 0;
 
     if (check_obj_lock) {
-      auto aiter = attrs.find(RGW_ATTR_OBJECT_RETENTION);
-      if (aiter != attrs.end()) {
-        RGWObjectRetention obj_retention;
-        try {
-          decode(obj_retention, aiter->second);
-        } catch (buffer::error& err) {
-          ldpp_dout(this, 0) << "ERROR: failed to decode RGWObjectRetention" << dendl;
-          op_ret = -EIO;
-          return;
-        }
-        if (ceph::real_clock::to_time_t(obj_retention.get_retain_until_date()) > ceph_clock_now()) {
-          if (obj_retention.get_mode().compare("GOVERNANCE") != 0 || !bypass_perm || !bypass_governance_mode) {
-            op_ret = -EACCES;
-            return;
-          }
-        }
-      }
-      aiter = attrs.find(RGW_ATTR_OBJECT_LEGAL_HOLD);
-      if (aiter != attrs.end()) {
-        RGWObjectLegalHold obj_legal_hold;
-        try {
-          decode(obj_legal_hold, aiter->second);
-        } catch (buffer::error& err) {
-          ldpp_dout(this, 0) << "ERROR: failed to decode RGWObjectLegalHold" << dendl;
-          op_ret = -EIO;
-          return;
-        }
-        if (obj_legal_hold.is_enabled()) {
-          op_ret = -EACCES;
-          return;
-        }
+      int object_lock_response = verify_object_lock(this, attrs, bypass_perm, bypass_governance_mode);
+      if (object_lock_response != 0) {
+        op_ret = object_lock_response;
+        return;
       }
     }
 
@@ -6551,6 +6524,27 @@ void RGWDeleteMultiObj::execute(optional_yield y)
              continue;
       }
     }
+
+    // verify_object_lock
+    bool check_obj_lock = obj->have_instance() && bucket->get_info().obj_lock_enabled();
+    if (check_obj_lock) {
+      int get_attrs_response = obj->get_obj_attrs(s->obj_ctx, s->yield);
+      if (get_attrs_response < 0) {
+        if (get_attrs_response == -ENOENT) {
+          // object maybe delete_marker, skip check_obj_lock
+          check_obj_lock = false;
+        } else {
+          // Something went wrong.
+          send_partial_response(*iter, false, "", get_attrs_response);
+          continue;
+        }
+      }
+    }
+    int object_lock_response = verify_object_lock(this, obj->get_attrs(), false, false);
+    if (object_lock_response != 0) {
+      send_partial_response(*iter, false, "", object_lock_response);
+      continue;
+    }
     
     // make reservation for notification if needed
     const auto versioned_object = s->bucket->versioning_enabled();