rgw_cksum: implement POST upload checksums

* properly transform pseudo headers in PostObj
* enable cksum verify in PostObj
* match checksum headers in match_policy_vars
* fixup add POST headers to environment

Signed-off-by: Matt Benjamin <[email protected]>

Author: mattbenjamin

src/rgw/rgw_cksum.h

@@ -206,7 +206,25 @@ namespace rgw { namespace cksum {
 	return ck.type;
     }
     return Type::none;
-  }
+  } /* parse_cksum_type */
+
+  static inline Type parse_cksum_type_hdr(const std::string_view hdr_name) {
+    auto pos = hdr_name.find("x-amz-checksum-", 0);
+    if (pos == std::string::npos) {
+      return Type::none;
+    }
+    constexpr int8_t psz = sizeof("x-amz-checksum-") - 1;
+    if ((hdr_name.size() - psz) > 0 ) {
+      std::string ck_name{hdr_name.substr(psz)};
+      return parse_cksum_type(ck_name.c_str());
+    }
+    return Type::none;
+  } /* parse_cksum_type_hdr */
+
+  static inline bool is_checksum_hdr(const std::string_view hdr_name) {
+    return hdr_name == "x-amz-checksum-algorithm" ||
+      parse_cksum_type_hdr(hdr_name) != Type::none;
+  } /* is_cksum_hdr */
 
   class Digest {
   public:

src/rgw/rgw_op.cc

@@ -4649,7 +4649,8 @@ void RGWPostObj::execute(optional_yield y)
 
   // make reservation for notification if needed
   std::unique_ptr<rgw::sal::Notification> res
-    = driver->get_notification(s->object.get(), s->src_object.get(), s, rgw::notify::ObjectCreatedPost, y);
+    = driver->get_notification(s->object.get(), s->src_object.get(), s,
+			       rgw::notify::ObjectCreatedPost, y);
   op_ret = res->publish_reserve(this);
   if (op_ret < 0) {
     return;
@@ -4700,10 +4701,13 @@ void RGWPostObj::execute(optional_yield y)
       return;
     }
 
+    std::unique_ptr<rgw::putobj::RGWPutObj_Cksum> cksum_filter;
+    std::unique_ptr<rgw::sal::DataProcessor> encrypt;
+
     /* No filters by default. */
     rgw::sal::DataProcessor *filter = processor.get();
 
-    std::unique_ptr<rgw::sal::DataProcessor> encrypt;
+    /* last filter runs first */
     op_ret = get_encrypt_filter(&encrypt, filter);
     if (op_ret < 0) {
       return;
@@ -4724,6 +4728,20 @@ void RGWPostObj::execute(optional_yield y)
       }
     }
 
+    /* XXX no lua filter? */
+
+    /* optional streaming checksum */
+    try {
+      cksum_filter =
+	rgw::putobj::RGWPutObj_Cksum::Factory(filter, *s->info.env);
+    } catch (const rgw::io::Exception& e) {
+      op_ret = e.code().value();
+      return;
+    }
+    if (cksum_filter) {
+      filter = &*cksum_filter;
+    }
+
     bool again;
     do {
       ceph::bufferlist data;
@@ -4738,6 +4756,7 @@ void RGWPostObj::execute(optional_yield y)
         break;
       }
 
+      /* XXXX we should modernize to use component buffers? */
       hash.Update((const unsigned char *)data.c_str(), data.length());
       op_ret = filter->process(std::move(data), ofs);
       if (op_ret < 0) {
@@ -4809,16 +4828,41 @@ void RGWPostObj::execute(optional_yield y)
       emplace_attr(RGW_ATTR_COMPRESSION, std::move(tmp));
     }
 
-    /* TODO:  implement POST checksums */
+    if (cksum_filter) {
+      auto cksum_verify =
+          cksum_filter->verify(*s->info.env); // valid or no supplied cksum
+      cksum = get<1>(cksum_verify);
+      if (std::get<0>(cksum_verify)) {
+        buffer::list cksum_bl;
+        cksum->encode(cksum_bl);
+        emplace_attr(RGW_ATTR_CKSUM, std::move(cksum_bl));
+      } else {
+        /* content checksum mismatch */
+        const auto &hdr = cksum_filter->header();
+        ldpp_dout(this, 4) << fmt::format("{} content checksum mismatch",
+                                          hdr.second)
+                           << fmt::format(
+                                  "\n\tcalculated={} != \n\texpected={}",
+                                  cksum->to_armor(),
+                                  cksum_filter->expected(*s->info.env))
+                           << dendl;
+        op_ret = -ERR_INVALID_REQUEST;
+        return;
+      }
+    }
+
     const req_context rctx{this, s->yield, s->trace.get()};
     op_ret = processor->complete(s->obj_size, etag, nullptr, real_time(),
-				 attrs, rgw::cksum::no_cksum,
+				 attrs, cksum,
 				 (delete_at ? *delete_at : real_time()),
 				 nullptr, nullptr, nullptr, nullptr, nullptr,
 				 rctx, rgw::sal::FLAG_LOG_OP);
     if (op_ret < 0) {
       return;
     }
+
+    /* XXX shouldn't we have an op-counter update here? */
+
   } while (is_next_file_to_upload());
 
   // send request to notification manager
@@ -4827,8 +4871,7 @@ void RGWPostObj::execute(optional_yield y)
     ldpp_dout(this, 1) << "ERROR: publishing notification failed, with error: " << ret << dendl;
     // too late to rollback operation, hence op_ret is not set here
   }
-}
-
+} /* RGWPostObj::execute() */
 
 void RGWPutMetadataAccount::filter_out_temp_url(map<string, bufferlist>& add_attrs,
                                                 const set<string>& rmattr_names,

src/rgw/rgw_op.h

@@ -1335,6 +1335,7 @@ class RGWPostObj : public RGWOp {
   RGWAccessControlPolicy policy;
   std::map<std::string, bufferlist> attrs;
   boost::optional<ceph::real_time> delete_at;
+  std::optional<rgw::cksum::Cksum> cksum;
 
   /* Must be called after get_data() or the result is undefined. */
   virtual std::string get_current_filename() const = 0;

src/rgw/rgw_policy_s3.cc

@@ -7,6 +7,7 @@
 #include "rgw_policy_s3.h"
 #include "rgw_common.h"
 #include "rgw_crypt_sanitize.h"
+#include "rgw_cksum.h"
 
 #define dout_context g_ceph_context
 #define dout_subsys ceph_subsys_rgw
@@ -101,15 +102,20 @@ bool RGWPolicyEnv::get_value(const string& s, string& val, map<string, bool, lts
   return get_var(var, val);
 }
 
-
-bool RGWPolicyEnv::match_policy_vars(map<string, bool, ltstr_nocase>& policy_vars, string& err_msg)
+bool RGWPolicyEnv::match_policy_vars(
+       map<string, bool, ltstr_nocase>& policy_vars, string& err_msg)
 {
   map<string, string, ltstr_nocase>::iterator iter;
   string ignore_prefix = "x-ignore-";
   for (iter = vars.begin(); iter != vars.end(); ++iter) {
     const string& var = iter->first;
-    if (strncasecmp(ignore_prefix.c_str(), var.c_str(), ignore_prefix.size()) == 0)
+    if (strncasecmp(ignore_prefix.c_str(), var.c_str(),
+		    ignore_prefix.size()) == 0) {
+      continue;
+    }
+    if (rgw::cksum::is_checksum_hdr(var)) {
       continue;
+    }
     if (policy_vars.count(var) == 0) {
       err_msg = "Policy missing condition: ";
       err_msg.append(iter->first);
@@ -118,7 +124,7 @@ bool RGWPolicyEnv::match_policy_vars(map<string, bool, ltstr_nocase>& policy_var
     }
   }
   return true;
-}
+} /* match_policy_vars */
 
 RGWPolicy::~RGWPolicy()
 {

src/rgw/rgw_rest_s3.cc

@@ -2962,22 +2962,33 @@ int RGWPostObj_ObjStore_S3::get_params(optional_yield y)
   } while (!done);
 
   for (auto &p: parts) {
-    if (! boost::istarts_with(p.first, "x-amz-server-side-encryption")) {
-      continue;
-    }
-    bufferlist &d { p.second.data };
-    std::string v { rgw_trim_whitespace(std::string_view(d.c_str(), d.length())) };
-    rgw_set_amz_meta_header(s->info.crypt_attribute_map, p.first, v, OVERWRITE);
-  }
+    if (boost::istarts_with(p.first, "x-amz-server-side-encryption")) {
+      bufferlist &d { p.second.data };
+      std::string v { rgw_trim_whitespace(std::string_view(d.c_str(), d.length())) };
+      rgw_set_amz_meta_header(s->info.crypt_attribute_map, p.first, v, OVERWRITE);
+    }
+    /* checksum headers */
+    auto& k = p.first;
+    auto cksum_type =  rgw::cksum::parse_cksum_type_hdr(k);
+    if (cksum_type != rgw::cksum::Type::none) {
+      put_prop("HTTP_X_AMZ_CHECKSUM_ALGORITHM",
+	       safe_upcase_str(to_string(cksum_type)));
+      bufferlist& d = p.second.data;
+      std::string v {
+	rgw_trim_whitespace(std::string_view(d.c_str(), d.length()))};
+      put_prop(ys_header_mangle(fmt::format("HTTP-{}", k)), v);
+    }
+  } /* each part */
 
   int r = get_encryption_defaults(s);
   if (r < 0) {
-    ldpp_dout(this, 5) << __func__ << "(): get_encryption_defaults() returned ret=" << r << dendl;
+    ldpp_dout(this, 5)
+      << __func__ << "(): get_encryption_defaults() returned ret=" << r << dendl;
     return r;
   }
 
   ldpp_dout(this, 20) << "adding bucket to policy env: " << s->bucket->get_name()
-		    << dendl;
+		      << dendl;
   env.add_var("bucket", s->bucket->get_name());
 
   string object_str;
@@ -3010,7 +3021,8 @@ int RGWPostObj_ObjStore_S3::get_params(optional_yield y)
   if (! storage_class.empty()) {
     s->dest_placement.storage_class = storage_class;
     if (!driver->valid_placement(s->dest_placement)) {
-      ldpp_dout(this, 0) << "NOTICE: invalid dest placement: " << s->dest_placement.to_str() << dendl;
+      ldpp_dout(this, 0) << "NOTICE: invalid dest placement: "
+			 << s->dest_placement.to_str() << dendl;
       err_msg = "The storage class you specified is not valid";
       return -EINVAL;
     }
@@ -3060,14 +3072,11 @@ int RGWPostObj_ObjStore_S3::get_params(optional_yield y)
   if (r < 0)
     return r;
 
-
   min_len = post_policy.min_length;
   max_len = post_policy.max_length;
 
-
-
   return 0;
-}
+} /* RGWPostObj_Objstore_S3::get_params() */
 
 int RGWPostObj_ObjStore_S3::get_tags()
 {

src/rgw/rgw_rest_s3.h

@@ -303,6 +303,12 @@ class RGWPostObj_ObjStore_S3 : public RGWPostObj_ObjStore {
   std::string get_current_filename() const override;
   std::string get_current_content_type() const override;
 
+  inline void put_prop(const std::string_view k, const std::string_view v) {
+    /* assume the caller will mangle the key name, if required */
+    auto& map = const_cast<env_map_t&>(s->info.env->get_map());
+    map.insert(env_map_t::value_type(k, v));
+  }
+
 public:
   RGWPostObj_ObjStore_S3() {}
   ~RGWPostObj_ObjStore_S3() override {}