Add a configurable to allow bucket perms to be checked before key perms

through rgw_defer_to_bucket_acls config option.  This configurable defaults
to an empty string.  Option values include:

  - recurse: If requesting perm PERM on a key, allow if user has
    PERM on the
    bucket to which the key belongs.

  - full_control: If requesting perm PERM on a key, allow if user
    has
    FULL_CONTROL on the bucket to which the key belongs.

This allows users to give someone full bucket perms and be able to
operate on the keys in the bucket without modifying the perms of every
key in the bucket.  This breaks S3 compatability, but that's why
it's a configurable!

Signed-off-by: Liam Monahan <[email protected]>

src/common/config_opts.h

@@ -706,6 +706,7 @@ OPTION(rgw_exit_timeout_secs, OPT_INT, 120) // how many seconds to wait for proc
 OPTION(rgw_get_obj_window_size, OPT_INT, 16 << 20) // window size in bytes for single get obj request
 OPTION(rgw_get_obj_max_req_size, OPT_INT, 4 << 20) // max length of a single get obj rados op
 OPTION(rgw_relaxed_s3_bucket_names, OPT_BOOL, false) // enable relaxed bucket name rules for US region buckets
+OPTION(rgw_defer_to_bucket_acls, OPT_STR, "") // if the user has bucket perms, use those before key perms (recurse and full_control)
 OPTION(rgw_list_buckets_max_chunk, OPT_INT, 1000) // max buckets to retrieve in a single op when listing user buckets
 OPTION(rgw_md_log_max_shards, OPT_INT, 64) // max shards for metadata log
 OPTION(rgw_num_zone_opstate_shards, OPT_INT, 128) // max shards for keeping inter-region copy progress info

src/rgw/rgw_common.cc

@@ -128,6 +128,7 @@ req_state::req_state(CephContext *_cct, class RGWEnv *e) : cct(_cct), cio(NULL),
 {
   enable_ops_log = e->conf->enable_ops_log;
   enable_usage_log = e->conf->enable_usage_log;
+  defer_to_bucket_acls = e->conf->defer_to_bucket_acls;
   content_started = false;
   format = 0;
   formatter = NULL;
@@ -618,8 +619,18 @@ bool verify_bucket_permission(struct req_state *s, int perm)
   return s->bucket_acl->verify_permission(s->user.user_id, perm, perm);
 }
 
+static inline bool check_deferred_bucket_acl(struct req_state *s, uint8_t deferred_check, int perm)
+{
+  return (s->defer_to_bucket_acls == deferred_check && verify_bucket_permission(s, perm));
+}
+
 bool verify_object_permission(struct req_state *s, RGWAccessControlPolicy *bucket_acl, RGWAccessControlPolicy *object_acl, int perm)
 {
+  if (check_deferred_bucket_acl(s, RGW_DEFER_TO_BUCKET_ACLS_RECURSE, perm) ||
+      check_deferred_bucket_acl(s, RGW_DEFER_TO_BUCKET_ACLS_FULL_CONTROL, RGW_PERM_FULL_CONTROL)) {
+    return true;
+  }
+
   if (!object_acl)
     return false;
 

src/rgw/rgw_common.h

@@ -94,6 +94,9 @@ using ceph::crypto::MD5;
 
 #define RGW_DEFAULT_MAX_BUCKETS 1000
 
+#define RGW_DEFER_TO_BUCKET_ACLS_RECURSE 1
+#define RGW_DEFER_TO_BUCKET_ACLS_FULL_CONTROL 2
+
 #define STATUS_CREATED           1900
 #define STATUS_ACCEPTED          1901
 #define STATUS_NO_CONTENT        1902
@@ -295,10 +298,11 @@ class RGWConf {
   void init(CephContext *cct, RGWEnv * env);
 public:
   RGWConf() :
-    enable_ops_log(1), enable_usage_log(1) {}
+    enable_ops_log(1), enable_usage_log(1), defer_to_bucket_acls(0) {}
 
   int enable_ops_log;
   int enable_usage_log;
+  uint8_t defer_to_bucket_acls;
 };
 
 enum http_op {
@@ -798,6 +802,7 @@ struct req_state {
    uint64_t obj_size;
    bool enable_ops_log;
    bool enable_usage_log;
+   uint8_t defer_to_bucket_acls;
    uint32_t perm_mask;
    utime_t header_time;
 

src/rgw/rgw_env.cc

@@ -108,4 +108,11 @@ void RGWConf::init(CephContext *cct, RGWEnv *env)
 {
   enable_ops_log = cct->_conf->rgw_enable_ops_log;
   enable_usage_log = cct->_conf->rgw_enable_usage_log;
+
+  defer_to_bucket_acls = 0;  // default
+  if (cct->_conf->rgw_defer_to_bucket_acls == "recurse") {
+    defer_to_bucket_acls = RGW_DEFER_TO_BUCKET_ACLS_RECURSE;
+  } else if (cct->_conf->rgw_defer_to_bucket_acls == "full_control") {
+    defer_to_bucket_acls = RGW_DEFER_TO_BUCKET_ACLS_FULL_CONTROL;
+  }
 }