ceph/src/rgw/rgw_user.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include <errno.h>
   5
   6 #include <string>
   7 #include <map>
   8 #include <boost/algorithm/string.hpp>
   9
  10 #include "common/errno.h"
  11 #include "common/Formatter.h"
  12 #include "common/ceph_json.h"
  13 #include "common/RWLock.h"
  14 #include "rgw_sal.h"
  15 #include "rgw_sal_rados.h"
  16 #include "rgw_zone.h"
  17 #include "rgw_acl.h"
  18
  19 #include "include/types.h"
  20 #include "rgw_user.h"
  21 #include "rgw_string.h"
  22
  23 // until everything is moved from rgw_common
  24 #include "rgw_common.h"
  25
  26 #include "rgw_bucket.h"
  27 #include "rgw_quota.h"
  28
  29 #include "services/svc_zone.h"
  30 #include "services/svc_sys_obj.h"
  31 #include "services/svc_sys_obj_cache.h"
  32 #include "services/svc_user.h"
  33 #include "services/svc_meta.h"
  34
  35 #define dout_subsys ceph_subsys_rgw
  36
  37
  38
  39 extern void op_type_to_str(uint32_t mask, char *buf, int len);
  40
  41 /**
  42  * Get the anonymous (ie, unauthenticated) user info.
  43  */
  44 void rgw_get_anon_user(RGWUserInfo& info)
  45 {
  46   info.user_id = RGW_USER_ANON_ID;
  47   info.display_name.clear();
  48   info.access_keys.clear();
  49 }
  50
  51 int rgw_user_sync_all_stats(const DoutPrefixProvider *dpp, rgw::sal::RGWRadosStore *store,
  52                             const rgw_user& user_id, optional_yield y)
  53 {
  54   rgw::sal::RGWBucketList user_buckets;
  55   rgw::sal::RGWRadosUser user(store, user_id);
  56
  57   CephContext *cct = store->ctx();
  58   size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
  59   string marker;
  60   int ret;
  61
  62   do {
  63     ret = user.list_buckets(dpp, marker, string(), max_entries, false, user_buckets, y);
  64     if (ret < 0) {
  65       ldpp_dout(dpp, 0) << "failed to read user buckets: ret=" << ret << dendl;
  66       return ret;
  67     }
  68     auto& buckets = user_buckets.get_buckets();
  69     for (auto i = buckets.begin(); i != buckets.end(); ++i) {
  70       marker = i->first;
  71
  72       auto& bucket = i->second;
  73
  74       ret = bucket->get_bucket_info(dpp, y);
  75       if (ret < 0) {
  76         ldpp_dout(dpp, 0) << "ERROR: could not read bucket info: bucket=" << bucket << " ret=" << ret << dendl;
  77         continue;
  78       }
  79       ret = bucket->sync_user_stats(dpp, y);
  80       if (ret < 0) {
  81         ldout(cct, 0) << "ERROR: could not sync bucket stats: ret=" << ret << dendl;
  82         return ret;
  83       }
  84       ret = bucket->check_bucket_shards(dpp);
  85       if (ret < 0) {
  86         ldpp_dout(dpp, 0) << "ERROR in check_bucket_shards: " << cpp_strerror(-ret)<< dendl;
  87       }
  88     }
  89   } while (user_buckets.is_truncated());
  90
  91   ret = store->ctl()->user->complete_flush_stats(dpp, user.get_user(), y);
  92   if (ret < 0) {
  93     cerr << "ERROR: failed to complete syncing user stats: ret=" << ret << std::endl;
  94     return ret;
  95   }
  96
  97   return 0;
  98 }
  99
 100 int rgw_user_get_all_buckets_stats(const DoutPrefixProvider *dpp,
 101                                    rgw::sal::RGWRadosStore *store,
 102                                    const rgw_user& user_id,
 103                                    map<string, cls_user_bucket_entry>& buckets_usage_map,
 104                                    optional_yield y)
 105 {
 106   CephContext *cct = store->ctx();
 107   size_t max_entries = cct->_conf->rgw_list_buckets_max_chunk;
 108   bool done;
 109   string marker;
 110   int ret;
 111
 112   do {
 113     rgw::sal::RGWBucketList buckets;
 114     ret = rgw_read_user_buckets(dpp, store, user_id, buckets, marker,
 115                                 string(), max_entries, false, y);
 116     if (ret < 0) {
 117       ldpp_dout(dpp, 0) << "failed to read user buckets: ret=" << ret << dendl;
 118       return ret;
 119     }
 120     auto& m = buckets.get_buckets();
 121     for (const auto& i :  m) {
 122       marker = i.first;
 123
 124       auto& bucket_ent = i.second;
 125       ret = bucket_ent->read_bucket_stats(dpp, y);
 126       if (ret < 0) {
 127         ldpp_dout(dpp, 0) << "ERROR: could not get bucket stats: ret=" << ret << dendl;
 128         return ret;
 129       }
 130       cls_user_bucket_entry entry;
 131       bucket_ent->convert(&entry);
 132       buckets_usage_map.emplace(bucket_ent->get_name(), entry);
 133     }
 134     done = (buckets.count() < max_entries);
 135   } while (!done);
 136
 137   return 0;
 138 }
 139
 140 /**
 141  * Save the given user information to storage.
 142  * Returns: 0 on success, -ERR# on failure.
 143  */
 144 int rgw_store_user_info(const DoutPrefixProvider *dpp,
 145                         RGWUserCtl *user_ctl,
 146                         RGWUserInfo& info,
 147                         RGWUserInfo *old_info,
 148                         RGWObjVersionTracker *objv_tracker,
 149                         real_time mtime,
 150                         bool exclusive,
 151                         optional_yield y,
 152                         map<string, bufferlist> *pattrs)
 153 {
 154   return user_ctl->store_info(dpp, info, y,
 155                               RGWUserCtl::PutParams()
 156                               .set_old_info(old_info)
 157                               .set_objv_tracker(objv_tracker)
 158                               .set_mtime(mtime)
 159                               .set_exclusive(exclusive)
 160                               .set_attrs(pattrs));
 161 }
 162
 163 /**
 164  * Given a uid, finds the user info associated with it.
 165  * returns: 0 on success, -ERR# on failure (including nonexistence)
 166  */
 167 int rgw_get_user_info_by_uid(const DoutPrefixProvider *dpp,
 168                              RGWUserCtl *user_ctl,
 169                              const rgw_user& uid,
 170                              RGWUserInfo& info,
 171                              optional_yield y,
 172                              RGWObjVersionTracker * const objv_tracker,
 173                              real_time * const pmtime,
 174                              rgw_cache_entry_info * const cache_info,
 175                              map<string, bufferlist> * const pattrs)
 176 {
 177   return user_ctl->get_info_by_uid(dpp, uid, &info, y,
 178                                    RGWUserCtl::GetParams()
 179                                    .set_objv_tracker(objv_tracker)
 180                                    .set_mtime(pmtime)
 181                                    .set_cache_info(cache_info)
 182                                    .set_attrs(pattrs));
 183 }
 184
 185 /**
 186  * Given an email, finds the user info associated with it.
 187  * returns: 0 on success, -ERR# on failure (including nonexistence)
 188  */
 189 int rgw_get_user_info_by_email(const DoutPrefixProvider *dpp,
 190                                RGWUserCtl *user_ctl, string& email,
 191                                RGWUserInfo& info, optional_yield y,
 192                                RGWObjVersionTracker *objv_tracker,
 193                                real_time *pmtime)
 194 {
 195   return user_ctl->get_info_by_email(dpp, email, &info, y,
 196                                      RGWUserCtl::GetParams()
 197                                      .set_objv_tracker(objv_tracker)
 198                                      .set_mtime(pmtime));
 199 }
 200
 201 /**
 202  * Given an swift username, finds the user_info associated with it.
 203  * returns: 0 on success, -ERR# on failure (including nonexistence)
 204  */
 205 int rgw_get_user_info_by_swift(const DoutPrefixProvider *dpp,
 206                                RGWUserCtl *user_ctl,
 207                                const string& swift_name,
 208                                RGWUserInfo& info,        /* out */
 209                                optional_yield y,
 210                                RGWObjVersionTracker * const objv_tracker,
 211                                real_time * const pmtime)
 212 {
 213   return user_ctl->get_info_by_swift(dpp, swift_name, &info, y,
 214                                      RGWUserCtl::GetParams()
 215                                      .set_objv_tracker(objv_tracker)
 216                                      .set_mtime(pmtime));
 217 }
 218
 219 /**
 220  * Given an access key, finds the user info associated with it.
 221  * returns: 0 on success, -ERR# on failure (including nonexistence)
 222  */
 223 extern int rgw_get_user_info_by_access_key(const DoutPrefixProvider *dpp,
 224                                            RGWUserCtl *user_ctl,
 225                                            const std::string& access_key,
 226                                            RGWUserInfo& info,
 227                                            optional_yield y,
 228                                            RGWObjVersionTracker* objv_tracker,
 229                                            real_time *pmtime)
 230 {
 231   return user_ctl->get_info_by_access_key(dpp, access_key, &info, y,
 232                                           RGWUserCtl::GetParams()
 233                                           .set_objv_tracker(objv_tracker)
 234                                           .set_mtime(pmtime));
 235 }
 236
 237 static bool char_is_unreserved_url(char c)
 238 {
 239   if (isalnum(c))
 240     return true;
 241
 242   switch (c) {
 243   case '-':
 244   case '.':
 245   case '_':
 246   case '~':
 247     return true;
 248   default:
 249     return false;
 250   }
 251 }
 252
 253 struct rgw_flags_desc {
 254   uint32_t mask;
 255   const char *str;
 256 };
 257
 258 static struct rgw_flags_desc rgw_perms[] = {
 259  { RGW_PERM_FULL_CONTROL, "full-control" },
 260  { RGW_PERM_READ | RGW_PERM_WRITE, "read-write" },
 261  { RGW_PERM_READ, "read" },
 262  { RGW_PERM_WRITE, "write" },
 263  { RGW_PERM_READ_ACP, "read-acp" },
 264  { RGW_PERM_WRITE_ACP, "write-acp" },
 265  { 0, NULL }
 266 };
 267
 268 void rgw_perm_to_str(uint32_t mask, char *buf, int len)
 269 {
 270   const char *sep = "";
 271   int pos = 0;
 272   if (!mask) {
 273     snprintf(buf, len, "<none>");
 274     return;
 275   }
 276   while (mask) {
 277     uint32_t orig_mask = mask;
 278     for (int i = 0; rgw_perms[i].mask; i++) {
 279       struct rgw_flags_desc *desc = &rgw_perms[i];
 280       if ((mask & desc->mask) == desc->mask) {
 281         pos += snprintf(buf + pos, len - pos, "%s%s", sep, desc->str);
 282         if (pos == len)
 283           return;
 284         sep = ", ";
 285         mask &= ~desc->mask;
 286         if (!mask)
 287           return;
 288       }
 289     }
 290     if (mask == orig_mask) // no change
 291       break;
 292   }
 293 }
 294
 295 uint32_t rgw_str_to_perm(const char *str)
 296 {
 297   if (strcasecmp(str, "") == 0)
 298     return RGW_PERM_NONE;
 299   else if (strcasecmp(str, "read") == 0)
 300     return RGW_PERM_READ;
 301   else if (strcasecmp(str, "write") == 0)
 302     return RGW_PERM_WRITE;
 303   else if (strcasecmp(str, "readwrite") == 0)
 304     return RGW_PERM_READ | RGW_PERM_WRITE;
 305   else if (strcasecmp(str, "full") == 0)
 306     return RGW_PERM_FULL_CONTROL;
 307
 308   return RGW_PERM_INVALID;
 309 }
 310
 311 int rgw_validate_tenant_name(const string& t)
 312 {
 313   struct tench {
 314     static bool is_good(char ch) {
 315       return isalnum(ch) || ch == '_';
 316     }
 317   };
 318   std::string::const_iterator it =
 319     std::find_if_not(t.begin(), t.end(), tench::is_good);
 320   return (it == t.end())? 0: -ERR_INVALID_TENANT_NAME;
 321 }
 322
 323 static bool validate_access_key(string& key)
 324 {
 325   const char *p = key.c_str();
 326   while (*p) {
 327     if (!char_is_unreserved_url(*p))
 328       return false;
 329     p++;
 330   }
 331   return true;
 332 }
 333
 334 static void set_err_msg(std::string *sink, std::string msg)
 335 {
 336   if (sink && !msg.empty())
 337     *sink = msg;
 338 }
 339
 340 /*
 341  * Dump either the full user info or a subset to a formatter.
 342  *
 343  * NOTE: It is the caller's respnsibility to ensure that the
 344  * formatter is flushed at the correct time.
 345  */
 346
 347 static void dump_subusers_info(Formatter *f, RGWUserInfo &info)
 348 {
 349   map<string, RGWSubUser>::iterator uiter;
 350
 351   f->open_array_section("subusers");
 352   for (uiter = info.subusers.begin(); uiter != info.subusers.end(); ++uiter) {
 353     RGWSubUser& u = uiter->second;
 354     f->open_object_section("user");
 355     string s;
 356     info.user_id.to_str(s);
 357     f->dump_format("id", "%s:%s", s.c_str(), u.name.c_str());
 358     char buf[256];
 359     rgw_perm_to_str(u.perm_mask, buf, sizeof(buf));
 360     f->dump_string("permissions", buf);
 361     f->close_section();
 362   }
 363   f->close_section();
 364 }
 365
 366 static void dump_access_keys_info(Formatter *f, RGWUserInfo &info)
 367 {
 368   map<string, RGWAccessKey>::iterator kiter;
 369   f->open_array_section("keys");
 370   for (kiter = info.access_keys.begin(); kiter != info.access_keys.end(); ++kiter) {
 371     RGWAccessKey& k = kiter->second;
 372     const char *sep = (k.subuser.empty() ? "" : ":");
 373     const char *subuser = (k.subuser.empty() ? "" : k.subuser.c_str());
 374     f->open_object_section("key");
 375     string s;
 376     info.user_id.to_str(s);
 377     f->dump_format("user", "%s%s%s", s.c_str(), sep, subuser);
 378     f->dump_string("access_key", k.id);
 379     f->dump_string("secret_key", k.key);
 380     f->close_section();
 381   }
 382   f->close_section();
 383 }
 384
 385 static void dump_swift_keys_info(Formatter *f, RGWUserInfo &info)
 386 {
 387   map<string, RGWAccessKey>::iterator kiter;
 388   f->open_array_section("swift_keys");
 389   for (kiter = info.swift_keys.begin(); kiter != info.swift_keys.end(); ++kiter) {
 390     RGWAccessKey& k = kiter->second;
 391     const char *sep = (k.subuser.empty() ? "" : ":");
 392     const char *subuser = (k.subuser.empty() ? "" : k.subuser.c_str());
 393     f->open_object_section("key");
 394     string s;
 395     info.user_id.to_str(s);
 396     f->dump_format("user", "%s%s%s", s.c_str(), sep, subuser);
 397     f->dump_string("secret_key", k.key);
 398     f->close_section();
 399   }
 400   f->close_section();
 401 }
 402
 403 static void dump_user_info(Formatter *f, RGWUserInfo &info,
 404                            RGWStorageStats *stats = NULL)
 405 {
 406   f->open_object_section("user_info");
 407   encode_json("tenant", info.user_id.tenant, f);
 408   encode_json("user_id", info.user_id.id, f);
 409   encode_json("display_name", info.display_name, f);
 410   encode_json("email", info.user_email, f);
 411   encode_json("suspended", (int)info.suspended, f);
 412   encode_json("max_buckets", (int)info.max_buckets, f);
 413
 414   dump_subusers_info(f, info);
 415   dump_access_keys_info(f, info);
 416   dump_swift_keys_info(f, info);
 417
 418   encode_json("caps", info.caps, f);
 419
 420   char buf[256];
 421   op_type_to_str(info.op_mask, buf, sizeof(buf));
 422   encode_json("op_mask", (const char *)buf, f);
 423   encode_json("system", (bool)info.system, f);
 424   encode_json("admin", (bool)info.admin, f);
 425   encode_json("default_placement", info.default_placement.name, f);
 426   encode_json("default_storage_class", info.default_placement.storage_class, f);
 427   encode_json("placement_tags", info.placement_tags, f);
 428   encode_json("bucket_quota", info.bucket_quota, f);
 429   encode_json("user_quota", info.user_quota, f);
 430   encode_json("temp_url_keys", info.temp_url_keys, f);
 431
 432   string user_source_type;
 433   switch ((RGWIdentityType)info.type) {
 434   case TYPE_RGW:
 435     user_source_type = "rgw";
 436     break;
 437   case TYPE_KEYSTONE:
 438     user_source_type = "keystone";
 439     break;
 440   case TYPE_LDAP:
 441     user_source_type = "ldap";
 442     break;
 443   case TYPE_NONE:
 444     user_source_type = "none";
 445     break;
 446   default:
 447     user_source_type = "none";
 448     break;
 449   }
 450   encode_json("type", user_source_type, f);
 451   encode_json("mfa_ids", info.mfa_ids, f);
 452   if (stats) {
 453     encode_json("stats", *stats, f);
 454   }
 455   f->close_section();
 456 }
 457
 458
 459 RGWAccessKeyPool::RGWAccessKeyPool(RGWUser* usr)
 460 {
 461   if (!usr) {
 462     return;
 463   }
 464
 465   user = usr;
 466
 467   store = user->get_store();
 468   user_ctl = user->get_user_ctl();
 469 }
 470
 471 int RGWAccessKeyPool::init(RGWUserAdminOpState& op_state)
 472 {
 473   if (!op_state.is_initialized()) {
 474     keys_allowed = false;
 475     return -EINVAL;
 476   }
 477
 478   rgw_user& uid = op_state.get_user_id();
 479   if (uid.compare(RGW_USER_ANON_ID) == 0) {
 480     keys_allowed = false;
 481     return -EACCES;
 482   }
 483
 484   swift_keys = op_state.get_swift_keys();
 485   access_keys = op_state.get_access_keys();
 486
 487   keys_allowed = true;
 488
 489   return 0;
 490 }
 491
 492 /*
 493  * Do a fairly exhaustive search for an existing key matching the parameters
 494  * given. Also handles the case where no key type was specified and updates
 495  * the operation state if needed.
 496  */
 497
 498 bool RGWAccessKeyPool::check_existing_key(RGWUserAdminOpState& op_state)
 499 {
 500   bool existing_key = false;
 501
 502   int key_type = op_state.get_key_type();
 503   std::string kid = op_state.get_access_key();
 504   std::map<std::string, RGWAccessKey>::iterator kiter;
 505   std::string swift_kid = op_state.build_default_swift_kid();
 506
 507   RGWUserInfo dup_info;
 508
 509   if (kid.empty() && swift_kid.empty())
 510     return false;
 511
 512   switch (key_type) {
 513   case KEY_TYPE_SWIFT:
 514     kiter = swift_keys->find(swift_kid);
 515
 516     existing_key = (kiter != swift_keys->end());
 517     if (existing_key)
 518       op_state.set_access_key(swift_kid);
 519
 520     break;
 521   case KEY_TYPE_S3:
 522     kiter = access_keys->find(kid);
 523     existing_key = (kiter != access_keys->end());
 524
 525     break;
 526   default:
 527     kiter = access_keys->find(kid);
 528
 529     existing_key = (kiter != access_keys->end());
 530     if (existing_key) {
 531       op_state.set_key_type(KEY_TYPE_S3);
 532       break;
 533     }
 534
 535     kiter = swift_keys->find(kid);
 536
 537     existing_key = (kiter != swift_keys->end());
 538     if (existing_key) {
 539       op_state.set_key_type(KEY_TYPE_SWIFT);
 540       break;
 541     }
 542
 543     // handle the case where the access key was not provided in user:key format
 544     if (swift_kid.empty())
 545       return false;
 546
 547     kiter = swift_keys->find(swift_kid);
 548
 549     existing_key = (kiter != swift_keys->end());
 550     if (existing_key) {
 551       op_state.set_access_key(swift_kid);
 552       op_state.set_key_type(KEY_TYPE_SWIFT);
 553     }
 554   }
 555
 556   op_state.set_existing_key(existing_key);
 557
 558   return existing_key;
 559 }
 560
 561 int RGWAccessKeyPool::check_op(RGWUserAdminOpState& op_state,
 562      std::string *err_msg)
 563 {
 564   RGWUserInfo dup_info;
 565
 566   if (!op_state.is_populated()) {
 567     set_err_msg(err_msg, "user info was not populated");
 568     return -EINVAL;
 569   }
 570
 571   if (!keys_allowed) {
 572     set_err_msg(err_msg, "keys not allowed for this user");
 573     return -EACCES;
 574   }
 575
 576   int32_t key_type = op_state.get_key_type();
 577
 578   // if a key type wasn't specified
 579   if (key_type < 0) {
 580       if (op_state.has_subuser()) {
 581         key_type = KEY_TYPE_SWIFT;
 582       } else {
 583         key_type = KEY_TYPE_S3;
 584       }
 585   }
 586
 587   op_state.set_key_type(key_type);
 588
 589   /* see if the access key was specified */
 590   if (key_type == KEY_TYPE_S3 && !op_state.will_gen_access() &&
 591       op_state.get_access_key().empty()) {
 592     set_err_msg(err_msg, "empty access key");
 593     return -ERR_INVALID_ACCESS_KEY;
 594   }
 595
 596   // don't check for secret key because we may be doing a removal
 597
 598   check_existing_key(op_state);
 599
 600   return 0;
 601 }
 602
 603 // Generate a new random key
 604 int RGWAccessKeyPool::generate_key(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state,
 605                                    optional_yield y, std::string *err_msg)
 606 {
 607   std::string id;
 608   std::string key;
 609
 610   std::pair<std::string, RGWAccessKey> key_pair;
 611   RGWAccessKey new_key;
 612   RGWUserInfo duplicate_check;
 613
 614   int key_type = op_state.get_key_type();
 615   bool gen_access = op_state.will_gen_access();
 616   bool gen_secret = op_state.will_gen_secret();
 617
 618   if (!keys_allowed) {
 619     set_err_msg(err_msg, "access keys not allowed for this user");
 620     return -EACCES;
 621   }
 622
 623   if (op_state.has_existing_key()) {
 624     set_err_msg(err_msg, "cannot create existing key");
 625     return -ERR_KEY_EXIST;
 626   }
 627
 628   if (!gen_access) {
 629     id = op_state.get_access_key();
 630   }
 631
 632   if (!id.empty()) {
 633     switch (key_type) {
 634     case KEY_TYPE_SWIFT:
 635       if (rgw_get_user_info_by_swift(dpp, user_ctl, id, duplicate_check, y) >= 0) {
 636         set_err_msg(err_msg, "existing swift key in RGW system:" + id);
 637         return -ERR_KEY_EXIST;
 638       }
 639       break;
 640     case KEY_TYPE_S3:
 641       if (rgw_get_user_info_by_access_key(dpp, user_ctl, id, duplicate_check, y) >= 0) {
 642         set_err_msg(err_msg, "existing S3 key in RGW system:" + id);
 643         return -ERR_KEY_EXIST;
 644       }
 645     }
 646   }
 647
 648   //key's subuser
 649   if (op_state.has_subuser()) {
 650     //create user and subuser at the same time, user's s3 key should not be set this
 651     if (!op_state.key_type_setbycontext || (key_type == KEY_TYPE_SWIFT)) {
 652       new_key.subuser = op_state.get_subuser();
 653     }
 654   }
 655
 656   //Secret key
 657   if (!gen_secret) {
 658     if (op_state.get_secret_key().empty()) {
 659       set_err_msg(err_msg, "empty secret key");
 660       return -ERR_INVALID_SECRET_KEY;
 661     }
 662
 663     key = op_state.get_secret_key();
 664   } else {
 665     char secret_key_buf[SECRET_KEY_LEN + 1];
 666     gen_rand_alphanumeric_plain(g_ceph_context, secret_key_buf, sizeof(secret_key_buf));
 667     key = secret_key_buf;
 668   }
 669
 670   // Generate the access key
 671   if (key_type == KEY_TYPE_S3 && gen_access) {
 672     char public_id_buf[PUBLIC_ID_LEN + 1];
 673
 674     do {
 675       int id_buf_size = sizeof(public_id_buf);
 676       gen_rand_alphanumeric_upper(g_ceph_context, public_id_buf, id_buf_size);
 677       id = public_id_buf;
 678       if (!validate_access_key(id))
 679         continue;
 680
 681     } while (!rgw_get_user_info_by_access_key(dpp, user_ctl, id, duplicate_check, y));
 682   }
 683
 684   if (key_type == KEY_TYPE_SWIFT) {
 685     id = op_state.build_default_swift_kid();
 686     if (id.empty()) {
 687       set_err_msg(err_msg, "empty swift access key");
 688       return -ERR_INVALID_ACCESS_KEY;
 689     }
 690
 691     // check that the access key doesn't exist
 692     if (rgw_get_user_info_by_swift(dpp, user_ctl, id, duplicate_check, y) >= 0) {
 693       set_err_msg(err_msg, "cannot create existing swift key");
 694       return -ERR_KEY_EXIST;
 695     }
 696   }
 697
 698   // finally create the new key
 699   new_key.id = id;
 700   new_key.key = key;
 701
 702   key_pair.first = id;
 703   key_pair.second = new_key;
 704
 705   if (key_type == KEY_TYPE_S3) {
 706     access_keys->insert(key_pair);
 707   } else if (key_type == KEY_TYPE_SWIFT) {
 708     swift_keys->insert(key_pair);
 709   }
 710
 711   return 0;
 712 }
 713
 714 // modify an existing key
 715 int RGWAccessKeyPool::modify_key(RGWUserAdminOpState& op_state, std::string *err_msg)
 716 {
 717   std::string id;
 718   std::string key = op_state.get_secret_key();
 719   int key_type = op_state.get_key_type();
 720
 721   RGWAccessKey modify_key;
 722
 723   pair<string, RGWAccessKey> key_pair;
 724   map<std::string, RGWAccessKey>::iterator kiter;
 725
 726   switch (key_type) {
 727   case KEY_TYPE_S3:
 728     id = op_state.get_access_key();
 729     if (id.empty()) {
 730       set_err_msg(err_msg, "no access key specified");
 731       return -ERR_INVALID_ACCESS_KEY;
 732     }
 733     break;
 734   case KEY_TYPE_SWIFT:
 735     id = op_state.build_default_swift_kid();
 736     if (id.empty()) {
 737       set_err_msg(err_msg, "no subuser specified");
 738       return -EINVAL;
 739     }
 740     break;
 741   default:
 742     set_err_msg(err_msg, "invalid key type");
 743     return -ERR_INVALID_KEY_TYPE;
 744   }
 745
 746   if (!op_state.has_existing_key()) {
 747     set_err_msg(err_msg, "key does not exist");
 748     return -ERR_INVALID_ACCESS_KEY;
 749   }
 750
 751   key_pair.first = id;
 752
 753   if (key_type == KEY_TYPE_SWIFT) {
 754     modify_key.id = id;
 755     modify_key.subuser = op_state.get_subuser();
 756   } else if (key_type == KEY_TYPE_S3) {
 757     kiter = access_keys->find(id);
 758     if (kiter != access_keys->end()) {
 759       modify_key = kiter->second;
 760     }
 761   }
 762
 763   if (op_state.will_gen_secret()) {
 764     char secret_key_buf[SECRET_KEY_LEN + 1];
 765     int key_buf_size = sizeof(secret_key_buf);
 766     gen_rand_alphanumeric_plain(g_ceph_context, secret_key_buf, key_buf_size);
 767     key = secret_key_buf;
 768   }
 769
 770   if (key.empty()) {
 771       set_err_msg(err_msg, "empty secret key");
 772       return -ERR_INVALID_SECRET_KEY;
 773   }
 774
 775   // update the access key with the new secret key
 776   modify_key.key = key;
 777
 778   key_pair.second = modify_key;
 779
 780
 781   if (key_type == KEY_TYPE_S3) {
 782     (*access_keys)[id] = modify_key;
 783   } else if (key_type == KEY_TYPE_SWIFT) {
 784     (*swift_keys)[id] = modify_key;
 785   }
 786
 787   return 0;
 788 }
 789
 790 int RGWAccessKeyPool::execute_add(const DoutPrefixProvider *dpp,
 791                                   RGWUserAdminOpState& op_state,
 792                                   std::string *err_msg, bool defer_user_update,
 793                                   optional_yield y)
 794 {
 795   int ret = 0;
 796
 797   std::string subprocess_msg;
 798   int key_op = GENERATE_KEY;
 799
 800   // set the op
 801   if (op_state.has_existing_key())
 802     key_op = MODIFY_KEY;
 803
 804   switch (key_op) {
 805   case GENERATE_KEY:
 806     ret = generate_key(dpp, op_state, y, &subprocess_msg);
 807     break;
 808   case MODIFY_KEY:
 809     ret = modify_key(op_state, &subprocess_msg);
 810     break;
 811   }
 812
 813   if (ret < 0) {
 814     set_err_msg(err_msg, subprocess_msg);
 815     return ret;
 816   }
 817
 818   // store the updated info
 819   if (!defer_user_update)
 820     ret = user->update(dpp, op_state, err_msg, y);
 821
 822   if (ret < 0)
 823     return ret;
 824
 825   return 0;
 826 }
 827
 828 int RGWAccessKeyPool::add(const DoutPrefixProvider *dpp,
 829                           RGWUserAdminOpState& op_state, optional_yield y,
 830                           std::string *err_msg)
 831 {
 832   return add(dpp, op_state, err_msg, false, y);
 833 }
 834
 835 int RGWAccessKeyPool::add(const DoutPrefixProvider *dpp,
 836                           RGWUserAdminOpState& op_state, std::string *err_msg,
 837                           bool defer_user_update, optional_yield y)
 838 {
 839   int ret;
 840   std::string subprocess_msg;
 841
 842   ret = check_op(op_state, &subprocess_msg);
 843   if (ret < 0) {
 844     set_err_msg(err_msg, "unable to parse request, " + subprocess_msg);
 845     return ret;
 846   }
 847
 848   ret = execute_add(dpp, op_state, &subprocess_msg, defer_user_update, y);
 849   if (ret < 0) {
 850     set_err_msg(err_msg, "unable to add access key, " + subprocess_msg);
 851     return ret;
 852   }
 853
 854   return 0;
 855 }
 856
 857 int RGWAccessKeyPool::execute_remove(const DoutPrefixProvider *dpp,
 858                                      RGWUserAdminOpState& op_state,
 859                                      std::string *err_msg,
 860                                      bool defer_user_update,
 861                                      optional_yield y)
 862 {
 863   int ret = 0;
 864
 865   int key_type = op_state.get_key_type();
 866   std::string id = op_state.get_access_key();
 867   map<std::string, RGWAccessKey>::iterator kiter;
 868   map<std::string, RGWAccessKey> *keys_map;
 869
 870   if (!op_state.has_existing_key()) {
 871     set_err_msg(err_msg, "unable to find access key");
 872     return -ERR_INVALID_ACCESS_KEY;
 873   }
 874
 875   if (key_type == KEY_TYPE_S3) {
 876     keys_map = access_keys;
 877   } else if (key_type == KEY_TYPE_SWIFT) {
 878     keys_map = swift_keys;
 879   } else {
 880     keys_map = NULL;
 881     set_err_msg(err_msg, "invalid access key");
 882     return -ERR_INVALID_ACCESS_KEY;
 883   }
 884
 885   kiter = keys_map->find(id);
 886   if (kiter == keys_map->end()) {
 887     set_err_msg(err_msg, "key not found");
 888     return -ERR_INVALID_ACCESS_KEY;
 889   }
 890
 891   keys_map->erase(kiter);
 892
 893   if (!defer_user_update)
 894     ret = user->update(dpp, op_state, err_msg, y);
 895
 896   if (ret < 0)
 897     return ret;
 898
 899   return 0;
 900 }
 901
 902 int RGWAccessKeyPool::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y,
 903                              std::string *err_msg)
 904 {
 905   return remove(dpp, op_state, err_msg, false, y);
 906 }
 907
 908 int RGWAccessKeyPool::remove(const DoutPrefixProvider *dpp,
 909                              RGWUserAdminOpState& op_state,
 910                              std::string *err_msg, bool defer_user_update,
 911                              optional_yield y)
 912 {
 913   int ret;
 914
 915   std::string subprocess_msg;
 916
 917   ret = check_op(op_state, &subprocess_msg);
 918   if (ret < 0) {
 919     set_err_msg(err_msg, "unable to parse request, " + subprocess_msg);
 920     return ret;
 921   }
 922
 923   ret = execute_remove(dpp, op_state, &subprocess_msg, defer_user_update, y);
 924   if (ret < 0) {
 925     set_err_msg(err_msg, "unable to remove access key, " + subprocess_msg);
 926     return ret;
 927   }
 928
 929   return 0;
 930 }
 931
 932 // remove all keys associated with a subuser
 933 int RGWAccessKeyPool::remove_subuser_keys(const DoutPrefixProvider *dpp,
 934                                           RGWUserAdminOpState& op_state,
 935                                           std::string *err_msg,
 936                                           bool defer_user_update,
 937                                           optional_yield y)
 938 {
 939   int ret = 0;
 940
 941   if (!op_state.is_populated()) {
 942     set_err_msg(err_msg, "user info was not populated");
 943     return -EINVAL;
 944   }
 945
 946   if (!op_state.has_subuser()) {
 947     set_err_msg(err_msg, "no subuser specified");
 948     return -EINVAL;
 949   }
 950
 951   std::string swift_kid = op_state.build_default_swift_kid();
 952   if (swift_kid.empty()) {
 953     set_err_msg(err_msg, "empty swift access key");
 954     return -EINVAL;
 955   }
 956
 957   map<std::string, RGWAccessKey>::iterator kiter;
 958   map<std::string, RGWAccessKey> *keys_map;
 959
 960   // a subuser can have at most one swift key
 961   keys_map = swift_keys;
 962   kiter = keys_map->find(swift_kid);
 963   if (kiter != keys_map->end()) {
 964     keys_map->erase(kiter);
 965   }
 966
 967   // a subuser may have multiple s3 key pairs
 968   std::string subuser_str = op_state.get_subuser();
 969   keys_map = access_keys;
 970   RGWUserInfo user_info = op_state.get_user_info();
 971   auto user_kiter = user_info.access_keys.begin();
 972   for (; user_kiter != user_info.access_keys.end(); ++user_kiter) {
 973     if (user_kiter->second.subuser == subuser_str) {
 974       kiter = keys_map->find(user_kiter->first);
 975       if (kiter != keys_map->end()) {
 976         keys_map->erase(kiter);
 977       }
 978     }
 979   }
 980
 981   if (!defer_user_update)
 982     ret = user->update(dpp, op_state, err_msg, y);
 983
 984   if (ret < 0)
 985     return ret;
 986
 987   return 0;
 988 }
 989
 990 RGWSubUserPool::RGWSubUserPool(RGWUser *usr)
 991 {
 992   if (!usr) {
 993     return;
 994   }
 995
 996   user = usr;
 997
 998   subusers_allowed = true;
 999   store = user->get_store();
1000   user_ctl = user->get_user_ctl();
1001 }
1002
1003 int RGWSubUserPool::init(RGWUserAdminOpState& op_state)
1004 {
1005   if (!op_state.is_initialized()) {
1006     subusers_allowed = false;
1007     return -EINVAL;
1008   }
1009
1010   rgw_user& uid = op_state.get_user_id();
1011   if (uid.compare(RGW_USER_ANON_ID) == 0) {
1012     subusers_allowed = false;
1013     return -EACCES;
1014   }
1015
1016   subuser_map = op_state.get_subusers();
1017   if (subuser_map == NULL) {
1018     subusers_allowed = false;
1019     return -EINVAL;
1020   }
1021
1022   subusers_allowed = true;
1023
1024   return 0;
1025 }
1026
1027 bool RGWSubUserPool::exists(std::string subuser)
1028 {
1029   if (subuser.empty())
1030     return false;
1031
1032   if (!subuser_map)
1033     return false;
1034
1035   if (subuser_map->count(subuser))
1036     return true;
1037
1038   return false;
1039 }
1040
1041 int RGWSubUserPool::check_op(RGWUserAdminOpState& op_state,
1042         std::string *err_msg)
1043 {
1044   bool existing = false;
1045   std::string subuser = op_state.get_subuser();
1046
1047   if (!op_state.is_populated()) {
1048     set_err_msg(err_msg, "user info was not populated");
1049     return -EINVAL;
1050   }
1051
1052   if (!subusers_allowed) {
1053     set_err_msg(err_msg, "subusers not allowed for this user");
1054     return -EACCES;
1055   }
1056
1057   if (subuser.empty() && !op_state.will_gen_subuser()) {
1058     set_err_msg(err_msg, "empty subuser name");
1059     return -EINVAL;
1060   }
1061
1062   if (op_state.get_subuser_perm() == RGW_PERM_INVALID) {
1063     set_err_msg(err_msg, "invaild subuser access");
1064     return -EINVAL;
1065   }
1066
1067   //set key type when it not set or set by context
1068   if ((op_state.get_key_type() < 0) || op_state.key_type_setbycontext) {
1069     op_state.set_key_type(KEY_TYPE_SWIFT);
1070     op_state.key_type_setbycontext = true;
1071   }
1072
1073   // check if the subuser exists
1074   if (!subuser.empty())
1075     existing = exists(subuser);
1076
1077   op_state.set_existing_subuser(existing);
1078
1079   return 0;
1080 }
1081
1082 int RGWSubUserPool::execute_add(const DoutPrefixProvider *dpp,
1083                                 RGWUserAdminOpState& op_state,
1084                                 std::string *err_msg, bool defer_user_update,
1085                                 optional_yield y)
1086 {
1087   int ret = 0;
1088   std::string subprocess_msg;
1089
1090   RGWSubUser subuser;
1091   std::pair<std::string, RGWSubUser> subuser_pair;
1092   std::string subuser_str = op_state.get_subuser();
1093
1094   subuser_pair.first = subuser_str;
1095
1096   // assumes key should be created
1097   if (op_state.has_key_op()) {
1098     ret = user->keys.add(dpp, op_state, &subprocess_msg, true, y);
1099     if (ret < 0) {
1100       set_err_msg(err_msg, "unable to create subuser key, " + subprocess_msg);
1101       return ret;
1102     }
1103   }
1104
1105   // create the subuser
1106   subuser.name = subuser_str;
1107
1108   if (op_state.has_subuser_perm())
1109     subuser.perm_mask = op_state.get_subuser_perm();
1110
1111   // insert the subuser into user info
1112   subuser_pair.second = subuser;
1113   subuser_map->insert(subuser_pair);
1114
1115   // attempt to save the subuser
1116   if (!defer_user_update)
1117     ret = user->update(dpp, op_state, err_msg, y);
1118
1119   if (ret < 0)
1120     return ret;
1121
1122   return 0;
1123 }
1124
1125 int RGWSubUserPool::add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y,
1126                         std::string *err_msg)
1127 {
1128   return add(dpp, op_state, err_msg, false, y);
1129 }
1130
1131 int RGWSubUserPool::add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg, bool defer_user_update, optional_yield y)
1132 {
1133   std::string subprocess_msg;
1134   int ret;
1135   int32_t key_type = op_state.get_key_type();
1136
1137   ret = check_op(op_state, &subprocess_msg);
1138   if (ret < 0) {
1139     set_err_msg(err_msg, "unable to parse request, " + subprocess_msg);
1140     return ret;
1141   }
1142
1143   if (key_type == KEY_TYPE_S3 && op_state.get_access_key().empty()) {
1144     op_state.set_gen_access();
1145   }
1146
1147   if (op_state.get_secret_key().empty()) {
1148     op_state.set_gen_secret();
1149   }
1150
1151   ret = execute_add(dpp, op_state, &subprocess_msg, defer_user_update, y);
1152   if (ret < 0) {
1153     set_err_msg(err_msg, "unable to create subuser, " + subprocess_msg);
1154     return ret;
1155   }
1156
1157   return 0;
1158 }
1159
1160 int RGWSubUserPool::execute_remove(const DoutPrefixProvider *dpp,
1161                                    RGWUserAdminOpState& op_state,
1162                                    std::string *err_msg, bool defer_user_update,
1163                                    optional_yield y)
1164 {
1165   int ret = 0;
1166   std::string subprocess_msg;
1167
1168   std::string subuser_str = op_state.get_subuser();
1169
1170   map<std::string, RGWSubUser>::iterator siter;
1171   siter = subuser_map->find(subuser_str);
1172   if (siter == subuser_map->end()){
1173     set_err_msg(err_msg, "subuser not found: " + subuser_str);
1174     return -ERR_NO_SUCH_SUBUSER;
1175   }
1176   if (!op_state.has_existing_subuser()) {
1177     set_err_msg(err_msg, "subuser not found: " + subuser_str);
1178     return -ERR_NO_SUCH_SUBUSER;
1179   }
1180
1181   // always purge all associate keys
1182   user->keys.remove_subuser_keys(dpp, op_state, &subprocess_msg, true, y);
1183
1184   // remove the subuser from the user info
1185   subuser_map->erase(siter);
1186
1187   // attempt to save the subuser
1188   if (!defer_user_update)
1189     ret = user->update(dpp, op_state, err_msg, y);
1190
1191   if (ret < 0)
1192     return ret;
1193
1194   return 0;
1195 }
1196
1197 int RGWSubUserPool::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y,
1198                            std::string *err_msg)
1199 {
1200   return remove(dpp, op_state, err_msg, false, y);
1201 }
1202
1203 int RGWSubUserPool::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg,
1204                            bool defer_user_update, optional_yield y)
1205 {
1206   std::string subprocess_msg;
1207   int ret;
1208
1209   ret = check_op(op_state, &subprocess_msg);
1210   if (ret < 0) {
1211     set_err_msg(err_msg, "unable to parse request, " + subprocess_msg);
1212     return ret;
1213   }
1214
1215   ret = execute_remove(dpp, op_state, &subprocess_msg, defer_user_update, y);
1216   if (ret < 0) {
1217     set_err_msg(err_msg, "unable to remove subuser, " + subprocess_msg);
1218     return ret;
1219   }
1220
1221   return 0;
1222 }
1223
1224 int RGWSubUserPool::execute_modify(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg, bool defer_user_update, optional_yield y)
1225 {
1226   int ret = 0;
1227   std::string subprocess_msg;
1228   std::map<std::string, RGWSubUser>::iterator siter;
1229   std::pair<std::string, RGWSubUser> subuser_pair;
1230
1231   std::string subuser_str = op_state.get_subuser();
1232   RGWSubUser subuser;
1233
1234   if (!op_state.has_existing_subuser()) {
1235     set_err_msg(err_msg, "subuser does not exist");
1236     return -ERR_NO_SUCH_SUBUSER;
1237   }
1238
1239   subuser_pair.first = subuser_str;
1240
1241   siter = subuser_map->find(subuser_str);
1242   subuser = siter->second;
1243
1244   if (op_state.has_key_op()) {
1245     ret = user->keys.add(dpp, op_state, &subprocess_msg, true, y);
1246     if (ret < 0) {
1247       set_err_msg(err_msg, "unable to create subuser keys, " + subprocess_msg);
1248       return ret;
1249     }
1250   }
1251
1252   if (op_state.has_subuser_perm())
1253     subuser.perm_mask = op_state.get_subuser_perm();
1254
1255   subuser_pair.second = subuser;
1256
1257   subuser_map->erase(siter);
1258   subuser_map->insert(subuser_pair);
1259
1260   // attempt to save the subuser
1261   if (!defer_user_update)
1262     ret = user->update(dpp, op_state, err_msg, y);
1263
1264   if (ret < 0)
1265     return ret;
1266
1267   return 0;
1268 }
1269
1270 int RGWSubUserPool::modify(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y, std::string *err_msg)
1271 {
1272   return RGWSubUserPool::modify(dpp, op_state, y, err_msg, false);
1273 }
1274
1275 int RGWSubUserPool::modify(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y, std::string *err_msg, bool defer_user_update)
1276 {
1277   std::string subprocess_msg;
1278   int ret;
1279
1280   RGWSubUser subuser;
1281
1282   ret = check_op(op_state, &subprocess_msg);
1283   if (ret < 0) {
1284     set_err_msg(err_msg, "unable to parse request, " + subprocess_msg);
1285     return ret;
1286   }
1287
1288   ret = execute_modify(dpp, op_state, &subprocess_msg, defer_user_update, y);
1289   if (ret < 0) {
1290     set_err_msg(err_msg, "unable to modify subuser, " + subprocess_msg);
1291     return ret;
1292   }
1293
1294   return 0;
1295 }
1296
1297 RGWUserCapPool::RGWUserCapPool(RGWUser *usr)
1298 {
1299   if (!usr) {
1300     return;
1301   }
1302   user = usr;
1303   caps_allowed = true;
1304 }
1305
1306 int RGWUserCapPool::init(RGWUserAdminOpState& op_state)
1307 {
1308   if (!op_state.is_initialized()) {
1309     caps_allowed = false;
1310     return -EINVAL;
1311   }
1312
1313   rgw_user& uid = op_state.get_user_id();
1314   if (uid.compare(RGW_USER_ANON_ID) == 0) {
1315     caps_allowed = false;
1316     return -EACCES;
1317   }
1318
1319   caps = op_state.get_caps_obj();
1320   if (!caps) {
1321     caps_allowed = false;
1322     return -ERR_INVALID_CAP;
1323   }
1324
1325   caps_allowed = true;
1326
1327   return 0;
1328 }
1329
1330 int RGWUserCapPool::add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y,
1331                         std::string *err_msg)
1332 {
1333   return add(dpp, op_state, err_msg, false, y);
1334 }
1335
1336 int RGWUserCapPool::add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg,
1337                         bool defer_save, optional_yield y)
1338 {
1339   int ret = 0;
1340   std::string caps_str = op_state.get_caps();
1341
1342   if (!op_state.is_populated()) {
1343     set_err_msg(err_msg, "user info was not populated");
1344     return -EINVAL;
1345   }
1346
1347   if (!caps_allowed) {
1348     set_err_msg(err_msg, "caps not allowed for this user");
1349     return -EACCES;
1350   }
1351
1352   if (caps_str.empty()) {
1353     set_err_msg(err_msg, "empty user caps");
1354     return -ERR_INVALID_CAP;
1355   }
1356
1357   int r = caps->add_from_string(caps_str);
1358   if (r < 0) {
1359     set_err_msg(err_msg, "unable to add caps: " + caps_str);
1360     return r;
1361   }
1362
1363   if (!defer_save)
1364     ret = user->update(dpp, op_state, err_msg, y);
1365
1366   if (ret < 0)
1367     return ret;
1368
1369   return 0;
1370 }
1371
1372 int RGWUserCapPool::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y,
1373                            std::string *err_msg)
1374 {
1375   return remove(dpp, op_state, err_msg, false, y);
1376 }
1377
1378 int RGWUserCapPool::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg,
1379                            bool defer_save, optional_yield y)
1380 {
1381   int ret = 0;
1382
1383   std::string caps_str = op_state.get_caps();
1384
1385   if (!op_state.is_populated()) {
1386     set_err_msg(err_msg, "user info was not populated");
1387     return -EINVAL;
1388   }
1389
1390   if (!caps_allowed) {
1391     set_err_msg(err_msg, "caps not allowed for this user");
1392     return -EACCES;
1393   }
1394
1395   if (caps_str.empty()) {
1396     set_err_msg(err_msg, "empty user caps");
1397     return -ERR_INVALID_CAP;
1398   }
1399
1400   int r = caps->remove_from_string(caps_str);
1401   if (r < 0) {
1402     set_err_msg(err_msg, "unable to remove caps: " + caps_str);
1403     return r;
1404   }
1405
1406   if (!defer_save)
1407     ret = user->update(dpp, op_state, err_msg, y);
1408
1409   if (ret < 0)
1410     return ret;
1411
1412   return 0;
1413 }
1414
1415 RGWUser::RGWUser() : caps(this), keys(this), subusers(this)
1416 {
1417   init_default();
1418 }
1419
1420 int RGWUser::init(const DoutPrefixProvider *dpp, rgw::sal::RGWRadosStore *storage,
1421                   RGWUserAdminOpState& op_state, optional_yield y)
1422 {
1423   init_default();
1424   int ret = init_storage(storage);
1425   if (ret < 0)
1426     return ret;
1427
1428   ret = init(dpp, op_state, y);
1429   if (ret < 0)
1430     return ret;
1431
1432   return 0;
1433 }
1434
1435 void RGWUser::init_default()
1436 {
1437   // use anonymous user info as a placeholder
1438   rgw_get_anon_user(old_info);
1439   user_id = RGW_USER_ANON_ID;
1440
1441   clear_populated();
1442 }
1443
1444 int RGWUser::init_storage(rgw::sal::RGWRadosStore *storage)
1445 {
1446   if (!storage) {
1447     return -EINVAL;
1448   }
1449
1450   store = storage;
1451   user_ctl = store->ctl()->user;
1452
1453   clear_populated();
1454
1455   /* API wrappers */
1456   keys = RGWAccessKeyPool(this);
1457   caps = RGWUserCapPool(this);
1458   subusers = RGWSubUserPool(this);
1459
1460   return 0;
1461 }
1462
1463 int RGWUser::init(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y)
1464 {
1465   bool found = false;
1466   std::string swift_user;
1467   user_id = op_state.get_user_id();
1468   std::string user_email = op_state.get_user_email();
1469   std::string access_key = op_state.get_access_key();
1470   std::string subuser = op_state.get_subuser();
1471
1472   int key_type = op_state.get_key_type();
1473   if (key_type == KEY_TYPE_SWIFT) {
1474     swift_user = op_state.get_access_key();
1475     access_key.clear();
1476   }
1477
1478   RGWUserInfo user_info;
1479
1480   clear_populated();
1481
1482   if (user_id.empty() && !subuser.empty()) {
1483     size_t pos = subuser.find(':');
1484     if (pos != string::npos) {
1485       user_id = subuser.substr(0, pos);
1486       op_state.set_user_id(user_id);
1487     }
1488   }
1489
1490   if (!user_id.empty() && (user_id.compare(RGW_USER_ANON_ID) != 0)) {
1491     found = (rgw_get_user_info_by_uid(dpp, user_ctl, user_id, user_info, y, &op_state.objv) >= 0);
1492     op_state.found_by_uid = found;
1493   }
1494   if (store->ctx()->_conf.get_val<bool>("rgw_user_unique_email")) {
1495     if (!user_email.empty() && !found) {
1496       found = (rgw_get_user_info_by_email(dpp, user_ctl, user_email, user_info, y, &op_state.objv) >= 0);
1497       op_state.found_by_email = found;
1498     }
1499   }
1500   if (!swift_user.empty() && !found) {
1501     found = (rgw_get_user_info_by_swift(dpp, user_ctl, swift_user, user_info, y, &op_state.objv) >= 0);
1502     op_state.found_by_key = found;
1503   }
1504   if (!access_key.empty() && !found) {
1505     found = (rgw_get_user_info_by_access_key(dpp, user_ctl, access_key, user_info, y, &op_state.objv) >= 0);
1506     op_state.found_by_key = found;
1507   }
1508
1509   op_state.set_existing_user(found);
1510   if (found) {
1511     op_state.set_user_info(user_info);
1512     op_state.set_populated();
1513
1514     old_info = user_info;
1515     set_populated();
1516   }
1517
1518   if (user_id.empty()) {
1519     user_id = user_info.user_id;
1520   }
1521   op_state.set_initialized();
1522
1523   // this may have been called by a helper object
1524   int ret = init_members(op_state);
1525   if (ret < 0)
1526     return ret;
1527
1528   return 0;
1529 }
1530
1531 int RGWUser::init_members(RGWUserAdminOpState& op_state)
1532 {
1533   int ret = 0;
1534
1535   ret = keys.init(op_state);
1536   if (ret < 0)
1537     return ret;
1538
1539   ret = subusers.init(op_state);
1540   if (ret < 0)
1541     return ret;
1542
1543   ret = caps.init(op_state);
1544   if (ret < 0)
1545     return ret;
1546
1547   return 0;
1548 }
1549
1550 int RGWUser::update(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg,
1551                     optional_yield y)
1552 {
1553   int ret;
1554   std::string subprocess_msg;
1555   RGWUserInfo user_info = op_state.get_user_info();
1556
1557   if (!store) {
1558     set_err_msg(err_msg, "couldn't initialize storage");
1559     return -EINVAL;
1560   }
1561
1562   RGWUserInfo *pold_info = (is_populated() ? &old_info : nullptr);
1563
1564   ret = rgw_store_user_info(dpp, user_ctl, user_info, pold_info, &op_state.objv,
1565                             real_time(), false, y);
1566   if (ret < 0) {
1567     set_err_msg(err_msg, "unable to store user info");
1568     return ret;
1569   }
1570
1571   old_info = user_info;
1572   set_populated();
1573
1574   return 0;
1575 }
1576
1577 int RGWUser::check_op(RGWUserAdminOpState& op_state, std::string *err_msg)
1578 {
1579   bool same_id;
1580   bool populated;
1581   rgw_user& op_id = op_state.get_user_id();
1582
1583   RGWUserInfo user_info;
1584
1585   same_id = (user_id.compare(op_id) == 0);
1586   populated = is_populated();
1587
1588   if (op_id.compare(RGW_USER_ANON_ID) == 0) {
1589     set_err_msg(err_msg, "unable to perform operations on the anonymous user");
1590     return -EINVAL;
1591   }
1592
1593   if (populated && !same_id) {
1594     set_err_msg(err_msg, "user id mismatch, operation id: " + op_id.to_str()
1595             + " does not match: " + user_id.to_str());
1596
1597     return -EINVAL;
1598   }
1599
1600   int ret = rgw_validate_tenant_name(op_id.tenant);
1601   if (ret) {
1602     set_err_msg(err_msg,
1603                 "invalid tenant only alphanumeric and _ characters are allowed");
1604     return ret;
1605   }
1606
1607   //set key type when it not set or set by context
1608   if ((op_state.get_key_type() < 0) || op_state.key_type_setbycontext) {
1609     op_state.set_key_type(KEY_TYPE_S3);
1610     op_state.key_type_setbycontext = true;
1611   }
1612
1613   return 0;
1614 }
1615
1616 // update swift_keys with new user id
1617 static void rename_swift_keys(const rgw_user& user,
1618                               std::map<std::string, RGWAccessKey>& keys)
1619 {
1620   std::string user_id;
1621   user.to_str(user_id);
1622
1623   auto modify_keys = std::move(keys);
1624   for ([[maybe_unused]] auto& [k, key] : modify_keys) {
1625     std::string id = user_id + ":" + key.subuser;
1626     key.id = id;
1627     keys[id] = std::move(key);
1628   }
1629 }
1630
1631 int RGWUser::execute_rename(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg, optional_yield y)
1632 {
1633   int ret;
1634   bool populated = op_state.is_populated();
1635
1636   if (!op_state.has_existing_user() && !populated) {
1637     set_err_msg(err_msg, "user not found");
1638     return -ENOENT;
1639   }
1640
1641   if (!populated) {
1642     ret = init(dpp, op_state, y);
1643     if (ret < 0) {
1644       set_err_msg(err_msg, "unable to retrieve user info");
1645       return ret;
1646     }
1647   }
1648
1649   rgw::sal::RGWRadosUser old_user(store, op_state.get_user_info());
1650   rgw::sal::RGWRadosUser new_user(store, op_state.get_new_uid());
1651   if (old_user.get_tenant() != new_user.get_tenant()) {
1652     set_err_msg(err_msg, "users have to be under the same tenant namespace "
1653                 + old_user.get_tenant() + " != " + new_user.get_tenant());
1654     return -EINVAL;
1655   }
1656
1657   // create a stub user and write only the uid index and buckets object
1658   RGWUserInfo stub_user_info;
1659   stub_user_info.user_id = new_user.get_user();
1660
1661   RGWObjVersionTracker objv;
1662   const bool exclusive = !op_state.get_overwrite_new_user(); // overwrite if requested
1663
1664   ret = user_ctl->store_info(dpp, stub_user_info, y,
1665                              RGWUserCtl::PutParams()
1666                              .set_objv_tracker(&objv)
1667                              .set_exclusive(exclusive));
1668   if (ret == -EEXIST) {
1669     set_err_msg(err_msg, "user name given by --new-uid already exists");
1670     return ret;
1671   }
1672   if (ret < 0) {
1673     set_err_msg(err_msg, "unable to store new user info");
1674     return ret;
1675   }
1676
1677   RGWAccessControlPolicy policy_instance;
1678   policy_instance.create_default(new_user.get_user(), old_user.get_display_name());
1679
1680   //unlink and link buckets to new user
1681   string marker;
1682   CephContext *cct = store->ctx();
1683   size_t max_buckets = cct->_conf->rgw_list_buckets_max_chunk;
1684   rgw::sal::RGWBucketList buckets;
1685
1686   do {
1687     ret = old_user.list_buckets(dpp, marker, "", max_buckets, false, buckets, y);
1688     if (ret < 0) {
1689       set_err_msg(err_msg, "unable to list user buckets");
1690       return ret;
1691     }
1692
1693     auto& m = buckets.get_buckets();
1694
1695     for (auto it = m.begin(); it != m.end(); ++it) {
1696       auto& bucket = it->second;
1697       marker = it->first;
1698
1699       ret = bucket->get_bucket_info(dpp, y);
1700       if (ret < 0) {
1701         set_err_msg(err_msg, "failed to fetch bucket info for bucket=" + bucket->get_name());
1702         return ret;
1703       }
1704
1705       ret = bucket->set_acl(dpp, policy_instance, y);
1706       if (ret < 0) {
1707         set_err_msg(err_msg, "failed to set acl on bucket " + bucket->get_name());
1708         return ret;
1709       }
1710
1711       ret = bucket->link(dpp, &new_user, y);
1712       if (ret < 0) {
1713         set_err_msg(err_msg, "failed to link bucket " + bucket->get_name());
1714         return ret;
1715       }
1716
1717       ret = bucket->chown(&new_user, &old_user, y, dpp);
1718       if (ret < 0) {
1719         set_err_msg(err_msg, "failed to run bucket chown" + cpp_strerror(-ret));
1720         return ret;
1721       }
1722     }
1723
1724   } while (buckets.is_truncated());
1725
1726   // update the 'stub user' with all of the other fields and rewrite all of the
1727   // associated index objects
1728   RGWUserInfo& user_info = op_state.get_user_info();
1729   user_info.user_id = new_user.get_user();
1730   op_state.objv = objv;
1731
1732   rename_swift_keys(new_user.get_user(), user_info.swift_keys);
1733
1734   return update(dpp, op_state, err_msg, y);
1735 }
1736
1737 int RGWUser::execute_add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg,
1738                          optional_yield y)
1739 {
1740   std::string subprocess_msg;
1741   int ret = 0;
1742   bool defer_user_update = true;
1743
1744   RGWUserInfo user_info;
1745
1746   rgw_user& uid = op_state.get_user_id();
1747   std::string user_email = op_state.get_user_email();
1748   std::string display_name = op_state.get_display_name();
1749
1750   // fail if the user exists already
1751   if (op_state.has_existing_user()) {
1752     if (op_state.found_by_email) {
1753       set_err_msg(err_msg, "email: " + user_email +
1754                   " is the email address an existing user");
1755       ret = -ERR_EMAIL_EXIST;
1756     } else if (op_state.found_by_key) {
1757       set_err_msg(err_msg, "duplicate key provided");
1758       ret = -ERR_KEY_EXIST;
1759     } else {
1760       set_err_msg(err_msg, "user: " + op_state.user_id.to_str() + " exists");
1761       ret = -EEXIST;
1762     }
1763     return ret;
1764   }
1765
1766   // fail if the user_info has already been populated
1767   if (op_state.is_populated()) {
1768     set_err_msg(err_msg, "cannot overwrite already populated user");
1769     return -EEXIST;
1770   }
1771
1772   // fail if the display name was not included
1773   if (display_name.empty()) {
1774     set_err_msg(err_msg, "no display name specified");
1775     return -EINVAL;
1776   }
1777
1778
1779   // set the user info
1780   user_id = uid;
1781   user_info.user_id = user_id;
1782   user_info.display_name = display_name;
1783   user_info.type = TYPE_RGW;
1784
1785   if (!user_email.empty())
1786     user_info.user_email = user_email;
1787
1788   CephContext *cct = store->ctx();
1789   if (op_state.max_buckets_specified) {
1790     user_info.max_buckets = op_state.get_max_buckets();
1791   } else {
1792     user_info.max_buckets =
1793       cct->_conf.get_val<int64_t>("rgw_user_max_buckets");
1794   }
1795
1796   user_info.suspended = op_state.get_suspension_status();
1797   user_info.admin = op_state.admin;
1798   user_info.system = op_state.system;
1799
1800   if (op_state.op_mask_specified)
1801     user_info.op_mask = op_state.get_op_mask();
1802
1803   if (op_state.has_bucket_quota()) {
1804     user_info.bucket_quota = op_state.get_bucket_quota();
1805   } else {
1806     rgw_apply_default_bucket_quota(user_info.bucket_quota, cct->_conf);
1807   }
1808
1809   if (op_state.temp_url_key_specified) {
1810     map<int, string>::iterator iter;
1811     for (iter = op_state.temp_url_keys.begin();
1812          iter != op_state.temp_url_keys.end(); ++iter) {
1813       user_info.temp_url_keys[iter->first] = iter->second;
1814     }
1815   }
1816
1817   if (op_state.has_user_quota()) {
1818     user_info.user_quota = op_state.get_user_quota();
1819   } else {
1820     rgw_apply_default_user_quota(user_info.user_quota, cct->_conf);
1821   }
1822
1823   if (op_state.default_placement_specified) {
1824     user_info.default_placement = op_state.default_placement;
1825   }
1826
1827   if (op_state.placement_tags_specified) {
1828     user_info.placement_tags = op_state.placement_tags;
1829   }
1830
1831   // update the request
1832   op_state.set_user_info(user_info);
1833   op_state.set_populated();
1834
1835   // update the helper objects
1836   ret = init_members(op_state);
1837   if (ret < 0) {
1838     set_err_msg(err_msg, "unable to initialize user");
1839     return ret;
1840   }
1841
1842   // see if we need to add an access key
1843   if (op_state.has_key_op()) {
1844     ret = keys.add(dpp, op_state, &subprocess_msg, defer_user_update, y);
1845     if (ret < 0) {
1846       set_err_msg(err_msg, "unable to create access key, " + subprocess_msg);
1847       return ret;
1848     }
1849   }
1850
1851   // see if we need to add some caps
1852   if (op_state.has_caps_op()) {
1853     ret = caps.add(dpp, op_state, &subprocess_msg, defer_user_update, y);
1854     if (ret < 0) {
1855       set_err_msg(err_msg, "unable to add user capabilities, " + subprocess_msg);
1856       return ret;
1857     }
1858   }
1859
1860   ret = update(dpp, op_state, err_msg, y);
1861   if (ret < 0)
1862     return ret;
1863
1864   return 0;
1865 }
1866
1867
1868 int RGWUser::add(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y, std::string *err_msg)
1869 {
1870   std::string subprocess_msg;
1871   int ret;
1872
1873   ret = check_op(op_state, &subprocess_msg);
1874   if (ret < 0) {
1875     set_err_msg(err_msg, "unable to parse parameters, " + subprocess_msg);
1876     return ret;
1877   }
1878
1879   ret = execute_add(dpp, op_state, &subprocess_msg, y);
1880   if (ret < 0) {
1881     set_err_msg(err_msg, "unable to create user, " + subprocess_msg);
1882     return ret;
1883   }
1884
1885   return 0;
1886 }
1887
1888 int RGWUser::rename(RGWUserAdminOpState& op_state, optional_yield y, const DoutPrefixProvider *dpp, std::string *err_msg)
1889 {
1890   std::string subprocess_msg;
1891   int ret;
1892
1893   ret = check_op(op_state, &subprocess_msg);
1894   if (ret < 0) {
1895     set_err_msg(err_msg, "unable to parse parameters, " + subprocess_msg);
1896     return ret;
1897   }
1898
1899   ret = execute_rename(dpp, op_state, &subprocess_msg, y);
1900   if (ret < 0) {
1901     set_err_msg(err_msg, "unable to rename user, " + subprocess_msg);
1902     return ret;
1903   }
1904
1905   return 0;
1906 }
1907
1908 int RGWUser::execute_remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg, optional_yield y)
1909 {
1910   int ret;
1911
1912   bool purge_data = op_state.will_purge_data();
1913   rgw_user& uid = op_state.get_user_id();
1914   RGWUserInfo user_info = op_state.get_user_info();
1915
1916   if (!op_state.has_existing_user()) {
1917     set_err_msg(err_msg, "user does not exist");
1918     return -ENOENT;
1919   }
1920
1921   rgw::sal::RGWBucketList buckets;
1922   string marker;
1923   CephContext *cct = store->ctx();
1924   size_t max_buckets = cct->_conf->rgw_list_buckets_max_chunk;
1925   do {
1926     ret = rgw_read_user_buckets(dpp, store, uid, buckets, marker, string(),
1927                                 max_buckets, false, y);
1928     if (ret < 0) {
1929       set_err_msg(err_msg, "unable to read user bucket info");
1930       return ret;
1931     }
1932
1933     auto& m = buckets.get_buckets();
1934     if (!m.empty() && !purge_data) {
1935       set_err_msg(err_msg, "must specify purge data to remove user with buckets");
1936       return -EEXIST; // change to code that maps to 409: conflict
1937     }
1938
1939     std::string prefix, delimiter;
1940     for (auto it = m.begin(); it != m.end(); ++it) {
1941       ret = it->second->remove_bucket(dpp, true, prefix, delimiter, false, nullptr, y);
1942       if (ret < 0) {
1943         set_err_msg(err_msg, "unable to delete user data");
1944         return ret;
1945       }
1946
1947       marker = it->first;
1948     }
1949
1950   } while (buckets.is_truncated());
1951
1952   ret = user_ctl->remove_info(dpp, user_info, y, RGWUserCtl::RemoveParams()
1953                                             .set_objv_tracker(&op_state.objv));
1954   if (ret < 0) {
1955     set_err_msg(err_msg, "unable to remove user from RADOS");
1956     return ret;
1957   }
1958
1959   op_state.clear_populated();
1960   clear_populated();
1961
1962   return 0;
1963 }
1964
1965 int RGWUser::remove(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y, std::string *err_msg)
1966 {
1967   std::string subprocess_msg;
1968   int ret;
1969
1970   ret = check_op(op_state, &subprocess_msg);
1971   if (ret < 0) {
1972     set_err_msg(err_msg, "unable to parse parameters, " + subprocess_msg);
1973     return ret;
1974   }
1975
1976   ret = execute_remove(dpp, op_state, &subprocess_msg, y);
1977   if (ret < 0) {
1978     set_err_msg(err_msg, "unable to remove user, " + subprocess_msg);
1979     return ret;
1980   }
1981
1982   return 0;
1983 }
1984
1985 int RGWUser::execute_modify(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, std::string *err_msg, optional_yield y)
1986 {
1987   bool populated = op_state.is_populated();
1988   int ret = 0;
1989   std::string subprocess_msg;
1990   std::string op_email = op_state.get_user_email();
1991   std::string display_name = op_state.get_display_name();
1992
1993   RGWUserInfo user_info;
1994   RGWUserInfo duplicate_check;
1995
1996   // ensure that the user info has been populated or is populate-able
1997   if (!op_state.has_existing_user() && !populated) {
1998     set_err_msg(err_msg, "user not found");
1999     return -ENOENT;
2000   }
2001
2002   // if the user hasn't already been populated...attempt to
2003   if (!populated) {
2004     ret = init(dpp, op_state, y);
2005     if (ret < 0) {
2006       set_err_msg(err_msg, "unable to retrieve user info");
2007       return ret;
2008     }
2009   }
2010
2011   // ensure that we can modify the user's attributes
2012   if (user_id.compare(RGW_USER_ANON_ID) == 0) {
2013     set_err_msg(err_msg, "unable to modify anonymous user's info");
2014     return -EACCES;
2015   }
2016
2017   user_info = old_info;
2018
2019   std::string old_email = old_info.user_email;
2020   if (!op_email.empty()) {
2021     // make sure we are not adding a duplicate email
2022     if (old_email != op_email) {
2023       ret = rgw_get_user_info_by_email(dpp, user_ctl, op_email, duplicate_check,y );
2024       if (ret >= 0 && duplicate_check.user_id.compare(user_id) != 0) {
2025         set_err_msg(err_msg, "cannot add duplicate email");
2026         return -ERR_EMAIL_EXIST;
2027       }
2028     }
2029     user_info.user_email = op_email;
2030   } else if (op_email.empty() && op_state.user_email_specified) {
2031     ldpp_dout(dpp, 10) << "removing email index: " << user_info.user_email << dendl;
2032     /* will be physically removed later when calling update() */
2033     user_info.user_email.clear();
2034   }
2035
2036   // update the remaining user info
2037   if (!display_name.empty())
2038     user_info.display_name = display_name;
2039
2040   if (op_state.max_buckets_specified)
2041     user_info.max_buckets = op_state.get_max_buckets();
2042
2043   if (op_state.admin_specified)
2044     user_info.admin = op_state.admin;
2045
2046   if (op_state.system_specified)
2047     user_info.system = op_state.system;
2048
2049   if (op_state.temp_url_key_specified) {
2050     map<int, string>::iterator iter;
2051     for (iter = op_state.temp_url_keys.begin();
2052          iter != op_state.temp_url_keys.end(); ++iter) {
2053       user_info.temp_url_keys[iter->first] = iter->second;
2054     }
2055   }
2056
2057   if (op_state.op_mask_specified)
2058     user_info.op_mask = op_state.get_op_mask();
2059
2060   if (op_state.has_bucket_quota())
2061     user_info.bucket_quota = op_state.get_bucket_quota();
2062
2063   if (op_state.has_user_quota())
2064     user_info.user_quota = op_state.get_user_quota();
2065
2066   if (op_state.has_suspension_op()) {
2067     __u8 suspended = op_state.get_suspension_status();
2068     user_info.suspended = suspended;
2069
2070     rgw::sal::RGWBucketList buckets;
2071
2072     if (user_id.empty()) {
2073       set_err_msg(err_msg, "empty user id passed...aborting");
2074       return -EINVAL;
2075     }
2076
2077     string marker;
2078     CephContext *cct = store->ctx();
2079     size_t max_buckets = cct->_conf->rgw_list_buckets_max_chunk;
2080     do {
2081       ret = rgw_read_user_buckets(dpp, store, user_id, buckets, marker, string(),
2082                                   max_buckets, false, y);
2083       if (ret < 0) {
2084         set_err_msg(err_msg, "could not get buckets for uid:  " + user_id.to_str());
2085         return ret;
2086       }
2087
2088       auto& m = buckets.get_buckets();
2089
2090       vector<rgw_bucket> bucket_names;
2091       for (auto iter = m.begin(); iter != m.end(); ++iter) {
2092         auto& bucket = iter->second;
2093         bucket_names.push_back(bucket->get_key());
2094
2095         marker = iter->first;
2096       }
2097
2098       ret = store->getRados()->set_buckets_enabled(bucket_names, !suspended, dpp);
2099       if (ret < 0) {
2100         set_err_msg(err_msg, "failed to modify bucket");
2101         return ret;
2102       }
2103
2104     } while (buckets.is_truncated());
2105   }
2106
2107   if (op_state.mfa_ids_specified) {
2108     user_info.mfa_ids = op_state.mfa_ids;
2109   }
2110
2111   if (op_state.default_placement_specified) {
2112     user_info.default_placement = op_state.default_placement;
2113   }
2114
2115   if (op_state.placement_tags_specified) {
2116     user_info.placement_tags = op_state.placement_tags;
2117   }
2118
2119   op_state.set_user_info(user_info);
2120
2121   // if we're supposed to modify keys, do so
2122   if (op_state.has_key_op()) {
2123     ret = keys.add(dpp, op_state, &subprocess_msg, true, y);
2124     if (ret < 0) {
2125       set_err_msg(err_msg, "unable to create or modify keys, " + subprocess_msg);
2126       return ret;
2127     }
2128   }
2129
2130   ret = update(dpp, op_state, err_msg, y);
2131   if (ret < 0)
2132     return ret;
2133
2134   return 0;
2135 }
2136
2137 int RGWUser::modify(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, optional_yield y, std::string *err_msg)
2138 {
2139   std::string subprocess_msg;
2140   int ret;
2141
2142   ret = check_op(op_state, &subprocess_msg);
2143   if (ret < 0) {
2144     if (is_populated() && (user_id.compare(op_state.get_user_id()) != 0)) {
2145       set_err_msg(err_msg, "unable to create user " + user_id.to_str()
2146                   + " because user id " + op_state.get_user_id().to_str()
2147                   + " already exists with email "
2148                   + op_state.get_user_email());
2149     } else {
2150       set_err_msg(err_msg, "unable to parse parameters, " + subprocess_msg);
2151     }
2152     return ret;
2153   }
2154
2155   ret = execute_modify(dpp, op_state, &subprocess_msg, y);
2156   if (ret < 0) {
2157     set_err_msg(err_msg, "unable to modify user, " + subprocess_msg);
2158     return ret;
2159   }
2160
2161   return 0;
2162 }
2163
2164 int RGWUser::info(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, RGWUserInfo& fetched_info,
2165                   optional_yield y, std::string *err_msg)
2166 {
2167   int ret = init(dpp, op_state, y);
2168   if (ret < 0) {
2169     set_err_msg(err_msg, "unable to fetch user info");
2170     return ret;
2171   }
2172
2173   fetched_info = op_state.get_user_info();
2174
2175   return 0;
2176 }
2177
2178 int RGWUser::info(RGWUserInfo& fetched_info, std::string *err_msg)
2179 {
2180   if (!is_populated()) {
2181     set_err_msg(err_msg, "no user info saved");
2182     return -EINVAL;
2183   }
2184
2185   fetched_info = old_info;
2186
2187   return 0;
2188 }
2189
2190 int RGWUser::list(const DoutPrefixProvider *dpp, RGWUserAdminOpState& op_state, RGWFormatterFlusher& flusher)
2191 {
2192   Formatter *formatter = flusher.get_formatter();
2193   void *handle = nullptr;
2194   std::string metadata_key = "user";
2195   if (op_state.max_entries > 1000) {
2196     op_state.max_entries = 1000;
2197   }
2198
2199   auto meta_mgr = store->ctl()->meta.mgr;
2200
2201   int ret = meta_mgr->list_keys_init(dpp, metadata_key, op_state.marker, &handle);
2202   if (ret < 0) {
2203     return ret;
2204   }
2205
2206   bool truncated = false;
2207   uint64_t count = 0;
2208   uint64_t left = 0;
2209   flusher.start(0);
2210
2211   // open the result object section
2212   formatter->open_object_section("result");
2213
2214   // open the user id list array section
2215   formatter->open_array_section("keys");
2216   do {
2217     std::list<std::string> keys;
2218     left = op_state.max_entries - count;
2219     ret = meta_mgr->list_keys_next(handle, left, keys, &truncated);
2220     if (ret < 0 && ret != -ENOENT) {
2221       return ret;
2222     } if (ret != -ENOENT) {
2223       for (std::list<std::string>::iterator iter = keys.begin(); iter != keys.end(); ++iter) {
2224       formatter->dump_string("key", *iter);
2225         ++count;
2226       }
2227     }
2228   } while (truncated && left > 0);
2229   // close user id list section
2230   formatter->close_section();
2231
2232   formatter->dump_bool("truncated", truncated);
2233   formatter->dump_int("count", count);
2234   if (truncated) {
2235     formatter->dump_string("marker", meta_mgr->get_marker(handle));
2236   }
2237
2238   // close result object section
2239   formatter->close_section();
2240
2241   meta_mgr->list_keys_complete(handle);
2242
2243   flusher.flush();
2244   return 0;
2245 }
2246
2247 int RGWUserAdminOp_User::list(const DoutPrefixProvider *dpp, rgw::sal::RGWRadosStore *store, RGWUserAdminOpState& op_state,
2248                   RGWFormatterFlusher& flusher)
2249 {
2250   RGWUser user;
2251
2252   int ret = user.init_storage(store);
2253   if (ret < 0)
2254     return ret;
2255
2256   ret = user.list(dpp, op_state, flusher);
2257   if (ret < 0)
2258     return ret;
2259
2260   return 0;
2261 }
2262
2263 int RGWUserAdminOp_User::info(const DoutPrefixProvider *dpp,
2264                               rgw::sal::RGWRadosStore *store, RGWUserAdminOpState& op_state,
2265                               RGWFormatterFlusher& flusher,
2266                               optional_yield y)
2267 {
2268   RGWUserInfo info;
2269   RGWUser user;
2270
2271   int ret = user.init(dpp, store, op_state, y);
2272   if (ret < 0)
2273     return ret;
2274
2275   if (!op_state.has_existing_user())
2276     return -ERR_NO_SUCH_USER;
2277
2278   Formatter *formatter = flusher.get_formatter();
2279
2280   ret = user.info(info, NULL);
2281   if (ret < 0)
2282     return ret;
2283
2284   if (op_state.sync_stats) {
2285     ret = rgw_user_sync_all_stats(dpp, store, info.user_id, y);
2286     if (ret < 0) {
2287       return ret;
2288     }
2289   }
2290
2291   RGWStorageStats stats;
2292   RGWStorageStats *arg_stats = NULL;
2293   if (op_state.fetch_stats) {
2294     int ret = store->ctl()->user->read_stats(dpp, info.user_id, &stats, y);
2295     if (ret < 0 && ret != -ENOENT) {
2296       return ret;
2297     }
2298
2299     arg_stats = &stats;
2300   }
2301
2302   if (formatter) {
2303     flusher.start(0);
2304
2305     dump_user_info(formatter, info, arg_stats);
2306     flusher.flush();
2307   }
2308
2309   return 0;
2310 }
2311
2312 int RGWUserAdminOp_User::create(const DoutPrefixProvider *dpp,
2313                                 rgw::sal::RGWRadosStore *store,
2314                                 RGWUserAdminOpState& op_state,
2315                                 RGWFormatterFlusher& flusher, optional_yield y)
2316 {
2317   RGWUserInfo info;
2318   RGWUser user;
2319   int ret = user.init(dpp, store, op_state, y);
2320   if (ret < 0)
2321     return ret;
2322
2323   Formatter *formatter = flusher.get_formatter();
2324
2325   ret = user.add(dpp, op_state, y, NULL);
2326   if (ret < 0) {
2327     if (ret == -EEXIST)
2328       ret = -ERR_USER_EXIST;
2329     return ret;
2330   }
2331
2332   ret = user.info(info, NULL);
2333   if (ret < 0)
2334     return ret;
2335
2336   if (formatter) {
2337     flusher.start(0);
2338
2339     dump_user_info(formatter, info);
2340     flusher.flush();
2341   }
2342
2343   return 0;
2344 }
2345
2346 int RGWUserAdminOp_User::modify(const DoutPrefixProvider *dpp,
2347                                 rgw::sal::RGWRadosStore *store,
2348                                 RGWUserAdminOpState& op_state,
2349                                 RGWFormatterFlusher& flusher, optional_yield y)
2350 {
2351   RGWUserInfo info;
2352   RGWUser user;
2353   int ret = user.init(dpp, store, op_state, y);
2354   if (ret < 0)
2355     return ret;
2356   Formatter *formatter = flusher.get_formatter();
2357
2358   ret = user.modify(dpp, op_state, y, NULL);
2359   if (ret < 0) {
2360     if (ret == -ENOENT)
2361       ret = -ERR_NO_SUCH_USER;
2362     return ret;
2363   }
2364
2365   ret = user.info(info, NULL);
2366   if (ret < 0)
2367     return ret;
2368
2369   if (formatter) {
2370     flusher.start(0);
2371
2372     dump_user_info(formatter, info);
2373     flusher.flush();
2374   }
2375
2376   return 0;
2377 }
2378
2379 int RGWUserAdminOp_User::remove(const DoutPrefixProvider *dpp, rgw::sal::RGWRadosStore *store, RGWUserAdminOpState& op_state,
2380                                 RGWFormatterFlusher& flusher, optional_yield y)
2381 {
2382   RGWUserInfo info;
2383   RGWUser user;
2384   int ret = user.init(dpp, store, op_state, y);
2385   if (ret < 0)
2386     return ret;
2387
2388
2389   ret = user.remove(dpp, op_state, y, NULL);
2390
2391   if (ret == -ENOENT)
2392     ret = -ERR_NO_SUCH_USER;
2393   return ret;
2394 }
2395
2396 int RGWUserAdminOp_Subuser::create(const DoutPrefixProvider *dpp,
2397                                    rgw::sal::RGWRadosStore *store,
2398                                    RGWUserAdminOpState& op_state,
2399                                    RGWFormatterFlusher& flusher,
2400                                    optional_yield y)
2401 {
2402   RGWUserInfo info;
2403   RGWUser user;
2404   int ret = user.init(dpp, store, op_state, y);
2405   if (ret < 0)
2406     return ret;
2407
2408   if (!op_state.has_existing_user())
2409     return -ERR_NO_SUCH_USER;
2410
2411   Formatter *formatter = flusher.get_formatter();
2412
2413   ret = user.subusers.add(dpp, op_state, y, NULL);
2414   if (ret < 0)
2415     return ret;
2416
2417   ret = user.info(info, NULL);
2418   if (ret < 0)
2419     return ret;
2420
2421   if (formatter) {
2422     flusher.start(0);
2423
2424     dump_subusers_info(formatter, info);
2425     flusher.flush();
2426   }
2427
2428   return 0;
2429 }
2430
2431 int RGWUserAdminOp_Subuser::modify(const DoutPrefixProvider *dpp,
2432                                    rgw::sal::RGWRadosStore *store, RGWUserAdminOpState& op_state,
2433                                    RGWFormatterFlusher& flusher, optional_yield y)
2434 {
2435   RGWUserInfo info;
2436   RGWUser user;
2437   int ret = user.init(dpp, store, op_state, y);
2438   if (ret < 0)
2439     return ret;
2440
2441   if (!op_state.has_existing_user())
2442     return -ERR_NO_SUCH_USER;
2443
2444   Formatter *formatter = flusher.get_formatter();
2445
2446   ret = user.subusers.modify(dpp, op_state, y, NULL);
2447   if (ret < 0)
2448     return ret;
2449
2450   ret = user.info(info, NULL);
2451   if (ret < 0)
2452     return ret;
2453
2454   if (formatter) {
2455     flusher.start(0);
2456
2457     dump_subusers_info(formatter, info);
2458     flusher.flush();
2459   }
2460
2461   return 0;
2462 }
2463
2464 int RGWUserAdminOp_Subuser::remove(const DoutPrefixProvider *dpp,
2465                                    rgw::sal::RGWRadosStore *store,
2466                                    RGWUserAdminOpState& op_state,
2467                                    RGWFormatterFlusher& flusher,
2468                                    optional_yield y)
2469 {
2470   RGWUserInfo info;
2471   RGWUser user;
2472   int ret = user.init(dpp, store, op_state, y);
2473   if (ret < 0)
2474     return ret;
2475
2476
2477   if (!op_state.has_existing_user())
2478     return -ERR_NO_SUCH_USER;
2479
2480   ret = user.subusers.remove(dpp, op_state, y, NULL);
2481   if (ret < 0)
2482     return ret;
2483
2484   return 0;
2485 }
2486
2487 int RGWUserAdminOp_Key::create(const DoutPrefixProvider *dpp,
2488                                rgw::sal::RGWRadosStore *store, RGWUserAdminOpState& op_state,
2489                                RGWFormatterFlusher& flusher,
2490                                optional_yield y)
2491 {
2492   RGWUserInfo info;
2493   RGWUser user;
2494   int ret = user.init(dpp, store, op_state, y);
2495   if (ret < 0)
2496     return ret;
2497
2498   if (!op_state.has_existing_user())
2499     return -ERR_NO_SUCH_USER;
2500
2501   Formatter *formatter = flusher.get_formatter();
2502
2503   ret = user.keys.add(dpp, op_state, y, NULL);
2504   if (ret < 0)
2505     return ret;
2506
2507   ret = user.info(info, NULL);
2508   if (ret < 0)
2509     return ret;
2510
2511   if (formatter) {
2512     flusher.start(0);
2513
2514     int key_type = op_state.get_key_type();
2515
2516     if (key_type == KEY_TYPE_SWIFT)
2517       dump_swift_keys_info(formatter, info);
2518
2519     else if (key_type == KEY_TYPE_S3)
2520       dump_access_keys_info(formatter, info);
2521
2522     flusher.flush();
2523   }
2524
2525   return 0;
2526 }
2527
2528 int RGWUserAdminOp_Key::remove(const DoutPrefixProvider *dpp,
2529                                rgw::sal::RGWRadosStore *store,
2530                                RGWUserAdminOpState& op_state,
2531                                RGWFormatterFlusher& flusher,
2532                                optional_yield y)
2533 {
2534   RGWUserInfo info;
2535   RGWUser user;
2536   int ret = user.init(dpp, store, op_state, y);
2537   if (ret < 0)
2538     return ret;
2539
2540   if (!op_state.has_existing_user())
2541     return -ERR_NO_SUCH_USER;
2542
2543
2544   ret = user.keys.remove(dpp, op_state, y, NULL);
2545   if (ret < 0)
2546     return ret;
2547
2548   return 0;
2549 }
2550
2551 int RGWUserAdminOp_Caps::add(const DoutPrefixProvider *dpp,
2552                              rgw::sal::RGWRadosStore *store,
2553                              RGWUserAdminOpState& op_state,
2554                              RGWFormatterFlusher& flusher, optional_yield y)
2555 {
2556   RGWUserInfo info;
2557   RGWUser user;
2558   int ret = user.init(dpp, store, op_state, y);
2559   if (ret < 0)
2560     return ret;
2561
2562   if (!op_state.has_existing_user())
2563     return -ERR_NO_SUCH_USER;
2564
2565   Formatter *formatter = flusher.get_formatter();
2566
2567   ret = user.caps.add(dpp, op_state, y, NULL);
2568   if (ret < 0)
2569     return ret;
2570
2571   ret = user.info(info, NULL);
2572   if (ret < 0)
2573     return ret;
2574
2575   if (formatter) {
2576     flusher.start(0);
2577
2578     info.caps.dump(formatter);
2579     flusher.flush();
2580   }
2581
2582   return 0;
2583 }
2584
2585
2586 int RGWUserAdminOp_Caps::remove(const DoutPrefixProvider *dpp,
2587                                 rgw::sal::RGWRadosStore *store,
2588                                 RGWUserAdminOpState& op_state,
2589                                 RGWFormatterFlusher& flusher, optional_yield y)
2590 {
2591   RGWUserInfo info;
2592   RGWUser user;
2593   int ret = user.init(dpp, store, op_state, y);
2594   if (ret < 0)
2595     return ret;
2596
2597   if (!op_state.has_existing_user())
2598     return -ERR_NO_SUCH_USER;
2599
2600   Formatter *formatter = flusher.get_formatter();
2601
2602   ret = user.caps.remove(dpp, op_state, y, NULL);
2603   if (ret < 0)
2604     return ret;
2605
2606   ret = user.info(info, NULL);
2607   if (ret < 0)
2608     return ret;
2609
2610   if (formatter) {
2611     flusher.start(0);
2612
2613     info.caps.dump(formatter);
2614     flusher.flush();
2615   }
2616
2617   return 0;
2618 }
2619
2620 class RGWUserMetadataHandler : public RGWMetadataHandler_GenericMetaBE {
2621 public:
2622   struct Svc {
2623     RGWSI_User *user{nullptr};
2624   } svc;
2625
2626   RGWUserMetadataHandler(RGWSI_User *user_svc) {
2627     base_init(user_svc->ctx(), user_svc->get_be_handler());
2628     svc.user = user_svc;
2629   }
2630
2631   string get_type() override { return "user"; }
2632
2633   int do_get(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWMetadataObject **obj, optional_yield y, const DoutPrefixProvider *dpp) override {
2634     RGWUserCompleteInfo uci;
2635     RGWObjVersionTracker objv_tracker;
2636     real_time mtime;
2637
2638     rgw_user user = RGWSI_User::user_from_meta_key(entry);
2639
2640     int ret = svc.user->read_user_info(op->ctx(), user, &uci.info, &objv_tracker,
2641                                        &mtime, nullptr, &uci.attrs,
2642                                        y, dpp);
2643     if (ret < 0) {
2644       return ret;
2645     }
2646
2647     RGWUserMetadataObject *mdo = new RGWUserMetadataObject(uci, objv_tracker.read_version, mtime);
2648     *obj = mdo;
2649
2650     return 0;
2651   }
2652
2653   RGWMetadataObject *get_meta_obj(JSONObj *jo, const obj_version& objv, const ceph::real_time& mtime) override {
2654     RGWUserCompleteInfo uci;
2655
2656     try {
2657       decode_json_obj(uci, jo);
2658     } catch (JSONDecoder::err& e) {
2659       return nullptr;
2660     }
2661
2662     return new RGWUserMetadataObject(uci, objv, mtime);
2663   }
2664
2665   int do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
2666              RGWMetadataObject *obj,
2667              RGWObjVersionTracker& objv_tracker,
2668              optional_yield y, const DoutPrefixProvider *dpp,
2669              RGWMDLogSyncType type, bool from_remote_zone) override;
2670
2671   int do_remove(RGWSI_MetaBackend_Handler::Op *op, string& entry, RGWObjVersionTracker& objv_tracker,
2672                 optional_yield y, const DoutPrefixProvider *dpp) override {
2673     RGWUserInfo info;
2674
2675     rgw_user user = RGWSI_User::user_from_meta_key(entry);
2676
2677     int ret = svc.user->read_user_info(op->ctx(), user, &info, nullptr,
2678                                        nullptr, nullptr, nullptr,
2679                                        y, dpp);
2680     if (ret < 0) {
2681       return ret;
2682     }
2683
2684     return svc.user->remove_user_info(op->ctx(), info, &objv_tracker,
2685                                       y, dpp);
2686   }
2687 };
2688
2689 class RGWMetadataHandlerPut_User : public RGWMetadataHandlerPut_SObj
2690 {
2691   RGWUserMetadataHandler *uhandler;
2692   RGWUserMetadataObject *uobj;
2693 public:
2694   RGWMetadataHandlerPut_User(RGWUserMetadataHandler *_handler,
2695                              RGWSI_MetaBackend_Handler::Op *op, string& entry,
2696                              RGWMetadataObject *obj, RGWObjVersionTracker& objv_tracker,
2697                              optional_yield y,
2698                              RGWMDLogSyncType type, bool from_remote_zone) : RGWMetadataHandlerPut_SObj(_handler, op, entry, obj, objv_tracker, y, type, from_remote_zone),
2699                                                                 uhandler(_handler) {
2700     uobj = static_cast<RGWUserMetadataObject *>(obj);
2701   }
2702
2703   int put_checked(const DoutPrefixProvider *dpp) override;
2704 };
2705
2706 int RGWUserMetadataHandler::do_put(RGWSI_MetaBackend_Handler::Op *op, string& entry,
2707                                    RGWMetadataObject *obj,
2708                                    RGWObjVersionTracker& objv_tracker,
2709                                    optional_yield y, const DoutPrefixProvider *dpp,
2710                                    RGWMDLogSyncType type, bool from_remote_zone)
2711 {
2712   RGWMetadataHandlerPut_User put_op(this, op, entry, obj, objv_tracker, y, type, from_remote_zone);
2713   return do_put_operate(&put_op, dpp);
2714 }
2715
2716 int RGWMetadataHandlerPut_User::put_checked(const DoutPrefixProvider *dpp)
2717 {
2718   RGWUserMetadataObject *orig_obj = static_cast<RGWUserMetadataObject *>(old_obj);
2719   RGWUserCompleteInfo& uci = uobj->get_uci();
2720
2721   map<string, bufferlist> *pattrs{nullptr};
2722   if (uci.has_attrs) {
2723     pattrs = &uci.attrs;
2724   }
2725
2726   RGWUserInfo *pold_info = (orig_obj ? &orig_obj->get_uci().info : nullptr);
2727
2728   auto mtime = obj->get_mtime();
2729
2730   int ret = uhandler->svc.user->store_user_info(op->ctx(), uci.info, pold_info,
2731                                                &objv_tracker, mtime,
2732                                                false, pattrs, y, dpp);
2733   if (ret < 0) {
2734     return ret;
2735   }
2736
2737   return STATUS_APPLIED;
2738 }
2739
2740
2741 RGWUserCtl::RGWUserCtl(RGWSI_Zone *zone_svc,
2742                        RGWSI_User *user_svc,
2743                        RGWUserMetadataHandler *_umhandler) : umhandler(_umhandler) {
2744   svc.zone = zone_svc;
2745   svc.user = user_svc;
2746   be_handler = umhandler->get_be_handler();
2747 }
2748
2749 template <class T>
2750 class optional_default
2751 {
2752   const std::optional<T>& opt;
2753   std::optional<T> def;
2754   const T *p;
2755 public:
2756   optional_default(const std::optional<T>& _o) : opt(_o) {
2757     if (opt) {
2758       p = &(*opt);
2759     } else {
2760       def = T();
2761       p = &(*def);
2762     }
2763   }
2764
2765   const T *operator->() {
2766     return p;
2767   }
2768
2769   const T& operator*() {
2770     return *p;
2771   }
2772 };
2773
2774 int RGWUserCtl::get_info_by_uid(const DoutPrefixProvider *dpp,
2775                                 const rgw_user& uid,
2776                                 RGWUserInfo *info,
2777                                 optional_yield y,
2778                                 const GetParams& params)
2779
2780 {
2781   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2782     return svc.user->read_user_info(op->ctx(),
2783                                     uid,
2784                                     info,
2785                                     params.objv_tracker,
2786                                     params.mtime,
2787                                     params.cache_info,
2788                                     params.attrs,
2789                                     y,
2790                                     dpp);
2791   });
2792 }
2793
2794 int RGWUserCtl::get_info_by_email(const DoutPrefixProvider *dpp,
2795                                   const string& email,
2796                                   RGWUserInfo *info,
2797                                   optional_yield y,
2798                                   const GetParams& params)
2799 {
2800   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2801     return svc.user->get_user_info_by_email(op->ctx(), email,
2802                                             info,
2803                                             params.objv_tracker,
2804                                             params.mtime,
2805                                             y,
2806                                             dpp);
2807   });
2808 }
2809
2810 int RGWUserCtl::get_info_by_swift(const DoutPrefixProvider *dpp,
2811                                   const string& swift_name,
2812                                   RGWUserInfo *info,
2813                                   optional_yield y,
2814                                   const GetParams& params)
2815 {
2816   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2817     return svc.user->get_user_info_by_swift(op->ctx(), swift_name,
2818                                             info,
2819                                             params.objv_tracker,
2820                                             params.mtime,
2821                                             y,
2822                                             dpp);
2823   });
2824 }
2825
2826 int RGWUserCtl::get_info_by_access_key(const DoutPrefixProvider *dpp,
2827                                        const string& access_key,
2828                                        RGWUserInfo *info,
2829                                        optional_yield y,
2830                                        const GetParams& params)
2831 {
2832   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2833     return svc.user->get_user_info_by_access_key(op->ctx(), access_key,
2834                                                  info,
2835                                                  params.objv_tracker,
2836                                                  params.mtime,
2837                                                  y,
2838                                                  dpp);
2839   });
2840 }
2841
2842 int RGWUserCtl::get_attrs_by_uid(const DoutPrefixProvider *dpp,
2843                                  const rgw_user& user_id,
2844                                  map<string, bufferlist> *pattrs,
2845                                  optional_yield y,
2846                                  RGWObjVersionTracker *objv_tracker)
2847 {
2848   RGWUserInfo user_info;
2849
2850   return get_info_by_uid(dpp, user_id, &user_info, y, RGWUserCtl::GetParams()
2851                          .set_attrs(pattrs)
2852                          .set_objv_tracker(objv_tracker));
2853 }
2854
2855 int RGWUserCtl::store_info(const DoutPrefixProvider *dpp,
2856                            const RGWUserInfo& info, optional_yield y,
2857                            const PutParams& params)
2858 {
2859   string key = RGWSI_User::get_meta_key(info.user_id);
2860
2861   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2862     return svc.user->store_user_info(op->ctx(), info,
2863                                      params.old_info,
2864                                      params.objv_tracker,
2865                                      params.mtime,
2866                                      params.exclusive,
2867                                      params.attrs,
2868                                      y,
2869                                      dpp);
2870   });
2871 }
2872
2873 int RGWUserCtl::remove_info(const DoutPrefixProvider *dpp,
2874                             const RGWUserInfo& info, optional_yield y,
2875                             const RemoveParams& params)
2876
2877 {
2878   string key = RGWSI_User::get_meta_key(info.user_id);
2879
2880   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2881     return svc.user->remove_user_info(op->ctx(), info,
2882                                       params.objv_tracker,
2883                                       y, dpp);
2884   });
2885 }
2886
2887 int RGWUserCtl::add_bucket(const DoutPrefixProvider *dpp,
2888                            const rgw_user& user,
2889                            const rgw_bucket& bucket,
2890                            ceph::real_time creation_time,
2891                            optional_yield y)
2892
2893 {
2894   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2895     return svc.user->add_bucket(dpp, op->ctx(), user, bucket, creation_time, y);
2896   });
2897 }
2898
2899 int RGWUserCtl::remove_bucket(const DoutPrefixProvider *dpp,
2900                               const rgw_user& user,
2901                               const rgw_bucket& bucket,
2902                               optional_yield y)
2903
2904 {
2905   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2906     return svc.user->remove_bucket(dpp, op->ctx(), user, bucket, y);
2907   });
2908 }
2909
2910 int RGWUserCtl::list_buckets(const DoutPrefixProvider *dpp,
2911                              const rgw_user& user,
2912                              const string& marker,
2913                              const string& end_marker,
2914                              uint64_t max,
2915                              bool need_stats,
2916                              RGWUserBuckets *buckets,
2917                              bool *is_truncated,
2918                              optional_yield y,
2919                              uint64_t default_max)
2920 {
2921   if (!max) {
2922     max = default_max;
2923   }
2924
2925   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2926     int ret = svc.user->list_buckets(dpp, op->ctx(), user, marker, end_marker,
2927                                      max, buckets, is_truncated, y);
2928     if (ret < 0) {
2929       return ret;
2930     }
2931     if (need_stats) {
2932       map<string, RGWBucketEnt>& m = buckets->get_buckets();
2933       ret = ctl.bucket->read_buckets_stats(m, y, dpp);
2934       if (ret < 0 && ret != -ENOENT) {
2935         ldpp_dout(dpp, 0) << "ERROR: could not get stats for buckets" << dendl;
2936         return ret;
2937       }
2938     }
2939     return 0;
2940   });
2941 }
2942
2943 int RGWUserCtl::flush_bucket_stats(const DoutPrefixProvider *dpp,
2944                                    const rgw_user& user,
2945                                    const RGWBucketEnt& ent,
2946                                    optional_yield y)
2947 {
2948   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2949     return svc.user->flush_bucket_stats(dpp, op->ctx(), user, ent, y);
2950   });
2951 }
2952
2953 int RGWUserCtl::complete_flush_stats(const DoutPrefixProvider *dpp, const rgw_user& user, optional_yield y)
2954 {
2955   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2956     return svc.user->complete_flush_stats(dpp, op->ctx(), user, y);
2957   });
2958 }
2959
2960 int RGWUserCtl::reset_stats(const DoutPrefixProvider *dpp, const rgw_user& user, optional_yield y)
2961 {
2962   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2963     return svc.user->reset_bucket_stats(dpp, op->ctx(), user, y);
2964   });
2965 }
2966
2967 int RGWUserCtl::read_stats(const DoutPrefixProvider *dpp,
2968                            const rgw_user& user, RGWStorageStats *stats,
2969                            optional_yield y,
2970                            ceph::real_time *last_stats_sync,
2971                            ceph::real_time *last_stats_update)
2972 {
2973   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2974     return svc.user->read_stats(dpp, op->ctx(), user, stats,
2975                                 last_stats_sync, last_stats_update, y);
2976   });
2977 }
2978
2979 int RGWUserCtl::read_stats_async(const DoutPrefixProvider *dpp, const rgw_user& user, RGWGetUserStats_CB *cb)
2980 {
2981   return be_handler->call([&](RGWSI_MetaBackend_Handler::Op *op) {
2982     return svc.user->read_stats_async(dpp, op->ctx(), user, cb);
2983   });
2984 }
2985
2986 RGWMetadataHandler *RGWUserMetaHandlerAllocator::alloc(RGWSI_User *user_svc) {
2987   return new RGWUserMetadataHandler(user_svc);
2988 }
2989