Merge pull request ceph#62284 from yuvalif/wip-yuval-70086

rgw/logging: use bucket policy for logging

Reviewed-by: Casey Bodley <[email protected]>

Author: cbodley

doc/radosgw/bucket_logging.rst

@@ -20,6 +20,12 @@ in different objects in the log bucket.
     - The log bucket must be created before enabling logging on a bucket
     - The log bucket cannot be the same as the bucket being logged
     - The log bucket cannot have logging enabled on it
+    - The log bucket cannot have any encryption set on in (including SSE-S3 with AES-256)
+    - The log bucket cannot have any compression set on it
+    - The log bucket must not have RequestPayer enabled
+    - Source and log bucket must be in the same zonegroup
+    - Source and log buckets may belong to different accounts (with proper bucket policy set)
+    - The log bucket may have object lock enabled with default retention period
 
 
 .. toctree::
@@ -51,6 +57,46 @@ Journal mode supports filtering out records based on matches of the prefixes and
 Note that it may happen that the log records were successfully written, but the bucket operation failed, since the logs are written.
 
 
+Bucket Logging Policy
+---------------------
+On the source bucket, only its owner is allowed to enable or disable bucket logging.
+For a bucket to be used as a log bucket, it must have bucket policy that allows that (even if the source bucket and the log bucket are owned by the same user or account).
+The bucket policy must allow the `s3:PutObject` action for the log bucket, to be perfomed by the `logging.s3.amazonaws.com` service principal.
+It should also specify the source bucket and account that are expected to write logs to it. For example:
+
+::
+
+    {
+      "Version": "2012-10-17",
+      "Statement": [
+        {
+          "Sid": "AllowLoggingFromSourceBucket",
+          "Effect": "Allow",
+          "Principal": {
+            "Service": "logging.s3.amazonaws.com"
+          },
+          "Action": "s3:PutObject",
+          "Resource": "arn:aws:s3:::log-bucket-name/prefix*",
+          "Condition": {
+            "StringEquals": {
+              "aws:SourceAccount": "source-account-id"
+            },
+            "ArnLike": {
+              "aws:SourceArn": "arn:aws:s3:::source-bucket-name"
+            }
+          }
+        }
+      ]
+    }
+
+
+Bucket Logging Quota
+--------------------
+Bucket and user quota are applied on the log bucket. Quota is checked every time a log record is written,
+and updated when the log object is added to the log bucket. In "Journal" mode, if the quota is exceeded, the logging operation will fail
+and as a result the bucket operation will also fail. In "Standard" mode, the logging operation will be skipped, but the bucket operation will continue.
+
+
 Bucket Logging REST API
 -----------------------
 Detailed under: `Bucket Operations`_.

src/rgw/driver/rados/rgw_sal_rados.cc

@@ -1072,7 +1072,7 @@ int RadosBucket::get_logging_object_name(std::string& obj_name,
   rgw_pool data_pool;
   const auto obj_name_oid = bucketlogging::object_name_oid(this, prefix);
   if (!store->getRados()->get_obj_data_pool(get_placement_rule(), rgw_obj{get_key(), obj_name_oid}, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
       "' when getting logging object name" << dendl;
     return -EIO;
   }
@@ -1088,6 +1088,10 @@ int RadosBucket::get_logging_object_name(std::string& obj_name,
                                nullptr,
                                nullptr);
   if (ret < 0) {
+    if (ret == -ENOENT) {
+      ldpp_dout(dpp, 20) << "INFO: logging object name '" << obj_name_oid << "' not found. ret = " << ret << dendl;
+      return ret;
+    }
     ldpp_dout(dpp, 1) << "ERROR: failed to get logging object name from '" << obj_name_oid << "'. ret = " << ret << dendl;
     return ret;
   }
@@ -1104,7 +1108,7 @@ int RadosBucket::set_logging_object_name(const std::string& obj_name,
   rgw_pool data_pool;
   const auto obj_name_oid = bucketlogging::object_name_oid(this, prefix);
   if (!store->getRados()->get_obj_data_pool(get_placement_rule(), rgw_obj{get_key(), obj_name_oid}, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
       "' when setting logging object name"  << dendl;
     return -EIO;
   }
@@ -1136,7 +1140,7 @@ int RadosBucket::remove_logging_object_name(const std::string& prefix,
   rgw_pool data_pool;
   const auto obj_name_oid = bucketlogging::object_name_oid(this, prefix);
   if (!store->getRados()->get_obj_data_pool(get_placement_rule(), rgw_obj{get_key(), obj_name_oid}, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
       "' when setting logging object name"  << dendl;
     return -EIO;
   }
@@ -1159,7 +1163,7 @@ int RadosBucket::remove_logging_object(const std::string& obj_name, optional_yie
   const auto placement_rule = get_placement_rule();
 
   if (!store->getRados()->get_obj_data_pool(placement_rule, head_obj, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
       "' when deleting logging object"  << dendl;
     return -EIO;
   }
@@ -1178,8 +1182,8 @@ int RadosBucket::commit_logging_object(const std::string& obj_name, optional_yie
   const auto placement_rule = get_placement_rule();
 
   if (!store->getRados()->get_obj_data_pool(placement_rule, head_obj, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
-      "' when comitting logging object"  << dendl;
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
+      "' when committing logging object"  << dendl;
     return -EIO;
   }
 
@@ -1197,7 +1201,7 @@ int RadosBucket::commit_logging_object(const std::string& obj_name, optional_yie
                      dpp,
                      &obj_attrs,
                      nullptr); ret < 0 && ret != -ENOENT) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to read logging data when comitting object '" << temp_obj_name
+    ldpp_dout(dpp, 1) << "ERROR: failed to read logging data when committing object '" << temp_obj_name
       << ". error: " << ret << dendl;
     return ret;
   } else if (ret == -ENOENT) {
@@ -1216,13 +1220,13 @@ int RadosBucket::commit_logging_object(const std::string& obj_name, optional_yie
                                 nullptr, // no special placment for tail
                                 get_key(),
                                 head_obj); ret < 0) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to create manifest when comitting logging object. error: " <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to create manifest when committing logging object. error: " <<
       ret << dendl;
     return ret;
   }
 
   if (const auto ret = manifest_gen.create_next(size); ret < 0) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to add object to manifest when comitting logging object. error: " <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to add object to manifest when committing logging object. error: " <<
       ret << dendl;
     return ret;
   }
@@ -1252,7 +1256,10 @@ int RadosBucket::commit_logging_object(const std::string& obj_name, optional_yie
   // TODO: head_obj_wop.meta.ptag
   // the owner of the logging object is the bucket owner
   // not the user that wrote the log that triggered the commit
-  const ACLOwner owner{bucket_info.owner, ""}; // TODO: missing display name
+  ACLOwner owner{bucket_info.owner, ""};
+  if (auto i = get_attrs().find(RGW_ATTR_ACL); i != get_attrs().end()) {
+    std::ignore = store->getRados()->decode_policy(dpp, i->second, &owner);
+  }
   head_obj_wop.meta.owner = owner;
   const auto etag = TOPNSPC::crypto::digest<TOPNSPC::crypto::MD5>(bl_data).to_str();
   bufferlist bl_etag;
@@ -1262,7 +1269,7 @@ int RadosBucket::commit_logging_object(const std::string& obj_name, optional_yie
   jspan_context trace{false, false};
   if (const auto ret = head_obj_wop.write_meta(0, size, obj_attrs, rctx, trace); ret < 0) {
   ldpp_dout(dpp, 1) << "ERROR: failed to commit logging object '" << temp_obj_name <<
-    "' to bucket id '" << get_info().bucket <<"'. error: " << ret << dendl;
+    "' to bucket '" << get_key() <<"'. error: " << ret << dendl;
     return ret;
   }
   ldpp_dout(dpp, 20) << "INFO: committed logging object '" << temp_obj_name <<
@@ -1300,7 +1307,7 @@ int RadosBucket::write_logging_object(const std::string& obj_name,
   rgw_pool data_pool;
   rgw_obj obj{get_key(), obj_name};
   if (!store->getRados()->get_obj_data_pool(get_placement_rule(), obj, &data_pool)) {
-    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_name() <<
+    ldpp_dout(dpp, 1) << "ERROR: failed to get data pool for bucket '" << get_key() <<
       "' when writing logging object" << dendl;
     return -EIO;
   }

src/rgw/rgw_auth.h

@@ -842,6 +842,67 @@ class RoleApplier : public IdentityApplier {
   };
 };
 
+class ServiceIdentity : public Identity {
+  const std::string service_id;
+public:
+  ServiceIdentity(const std::string& s) : service_id(s) {}
+  virtual ~ServiceIdentity() = default;
+
+  ACLOwner get_aclowner() const override {
+    return {};
+  }
+
+  uint32_t get_perms_from_aclspec(const DoutPrefixProvider* dpp, const aclspec_t& aclspec) const override {
+    return RGW_PERM_NONE;
+  }
+
+  bool is_admin_of(const rgw_owner& o) const override {
+    return false;
+  }
+
+  bool is_owner_of(const rgw_owner& o) const override {
+    return false;
+  }
+
+  uint32_t get_perm_mask() const override {
+    return RGW_PERM_NONE;
+  }
+
+  virtual void to_str(std::ostream& out) const override {
+    out << "rgw::auth::ServiceIdentity(id=" << service_id << ")";
+  }
+
+  bool is_identity(const Principal& p) const override {
+    return p.is_service() && p.get_service() == service_id;
+  }
+
+  uint32_t get_identity_type() const override {
+    return TYPE_RGW;
+  }
+
+  std::string get_acct_name() const override {
+    return {};
+  }
+
+  std::string get_subuser() const override {
+    return {};
+  }
+
+  const std::string& get_tenant() const override {
+    static const std::string no_tenant;
+    return no_tenant;
+  }
+
+  const std::optional<RGWAccountInfo>& get_account() const override {
+    static constexpr std::optional<RGWAccountInfo> no_account;
+    return no_account;
+  }
+
+  bool is_root() const override {
+    return false;
+  }
+};
+
 /* The anonymous abstract engine. */
 class AnonymousEngine : public Engine {
   CephContext* const cct;

src/rgw/rgw_basic_types.cc

@@ -170,6 +170,9 @@ ostream& operator <<(ostream& m, const Principal& p) {
   if (p.is_wildcard()) {
     return m << "*";
   }
+  if (p.is_service()) {
+    return m << p.get_service();
+  }
 
   m << "arn:aws:iam:" << p.get_account() << ":";
   if (p.is_account()) {

src/rgw/rgw_basic_types.h

@@ -143,10 +143,11 @@ extern void decode_json_obj(rgw_placement_rule& v, JSONObj *obj);
 namespace rgw {
 namespace auth {
 class Principal {
-  enum types { User, Role, Account, Wildcard, OidcProvider, AssumedRole };
+  enum types { User, Role, Account, Wildcard, OidcProvider, AssumedRole, Service };
   types t;
   rgw_user u;
   std::string idp_url;
+  std::string service_id;
 
   explicit Principal(types t)
     : t(t) {}
@@ -183,6 +184,12 @@ class Principal {
     return Principal(AssumedRole, std::move(t), std::move(u));
   }
 
+  static Principal service(std::string&& s) {
+    auto p = Principal(Service);
+    p.service_id = std::move(s);
+    return p;
+  }
+
   bool is_wildcard() const {
     return t == Wildcard;
   }
@@ -207,6 +214,10 @@ class Principal {
     return t == AssumedRole;
   }
 
+  bool is_service() const {
+    return t == Service;
+  }
+
   const std::string& get_account() const {
     return u.tenant;
   }
@@ -227,6 +238,10 @@ class Principal {
     return u.id;
   }
 
+  const std::string& get_service() const {
+    return service_id;
+  }
+
   bool operator ==(const Principal& o) const {
     return (t == o.t) && (u == o.u);
   }