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