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