ceph/src/pybind/mgr/dashboard/controllers/rgw.py

   1 # -*- coding: utf-8 -*-
   2
   3 import json
   4 import logging
   5
   6 import cherrypy
   7
   8 from ..exceptions import DashboardException
   9 from ..rest_client import RequestException
  10 from ..security import Permission, Scope
  11 from ..services.auth import AuthManager, JwtManager
  12 from ..services.ceph_service import CephService
  13 from ..services.rgw_client import NoRgwDaemonsException, RgwClient
  14 from ..tools import json_str_to_object, str_to_bool
  15 from . import APIDoc, APIRouter, BaseController, Endpoint, EndpointDoc, \
  16     ReadPermission, RESTController, UIRouter, allow_empty_body
  17 from ._version import APIVersion
  18
  19 try:
  20     from typing import Any, Dict, List, Optional, Union
  21 except ImportError:  # pragma: no cover
  22     pass  # Just for type checking
  23
  24 logger = logging.getLogger("controllers.rgw")
  25
  26 RGW_SCHEMA = {
  27     "available": (bool, "Is RGW available?"),
  28     "message": (str, "Descriptions")
  29 }
  30
  31 RGW_DAEMON_SCHEMA = {
  32     "id": (str, "Daemon ID"),
  33     "version": (str, "Ceph Version"),
  34     "server_hostname": (str, ""),
  35     "zonegroup_name": (str, "Zone Group"),
  36     "zone_name": (str, "Zone")
  37 }
  38
  39 RGW_USER_SCHEMA = {
  40     "list_of_users": ([str], "list of rgw users")
  41 }
  42
  43
  44 @UIRouter('/rgw', Scope.RGW)
  45 @APIDoc("RGW Management API", "Rgw")
  46 class Rgw(BaseController):
  47     @Endpoint()
  48     @ReadPermission
  49     @EndpointDoc("Display RGW Status",
  50                  responses={200: RGW_SCHEMA})
  51     def status(self) -> dict:
  52         status = {'available': False, 'message': None}
  53         try:
  54             instance = RgwClient.admin_instance()
  55             # Check if the service is online.
  56             try:
  57                 is_online = instance.is_service_online()
  58             except RequestException as e:
  59                 # Drop this instance because the RGW client seems not to
  60                 # exist anymore (maybe removed via orchestrator). Removing
  61                 # the instance from the cache will result in the correct
  62                 # error message next time when the backend tries to
  63                 # establish a new connection (-> 'No RGW found' instead
  64                 # of 'RGW REST API failed request ...').
  65                 # Note, this only applies to auto-detected RGW clients.
  66                 RgwClient.drop_instance(instance)
  67                 raise e
  68             if not is_online:
  69                 msg = 'Failed to connect to the Object Gateway\'s Admin Ops API.'
  70                 raise RequestException(msg)
  71             # Ensure the system flag is set for the API user ID.
  72             if not instance.is_system_user():  # pragma: no cover - no complexity there
  73                 msg = 'The system flag is not set for user "{}".'.format(
  74                     instance.userid)
  75                 raise RequestException(msg)
  76             status['available'] = True
  77         except (DashboardException, RequestException, NoRgwDaemonsException) as ex:
  78             status['message'] = str(ex)  # type: ignore
  79         return status
  80
  81
  82 @APIRouter('/rgw/daemon', Scope.RGW)
  83 @APIDoc("RGW Daemon Management API", "RgwDaemon")
  84 class RgwDaemon(RESTController):
  85     @EndpointDoc("Display RGW Daemons",
  86                  responses={200: [RGW_DAEMON_SCHEMA]})
  87     def list(self) -> List[dict]:
  88         daemons: List[dict] = []
  89         try:
  90             instance = RgwClient.admin_instance()
  91         except NoRgwDaemonsException:
  92             return daemons
  93
  94         for hostname, server in CephService.get_service_map('rgw').items():
  95             for service in server['services']:
  96                 metadata = service['metadata']
  97
  98                 # extract per-daemon service data and health
  99                 daemon = {
 100                     'id': metadata['id'],
 101                     'service_map_id': service['id'],
 102                     'version': metadata['ceph_version'],
 103                     'server_hostname': hostname,
 104                     'realm_name': metadata['realm_name'],
 105                     'zonegroup_name': metadata['zonegroup_name'],
 106                     'zone_name': metadata['zone_name'],
 107                     'default': instance.daemon.name == metadata['id']
 108                 }
 109
 110                 daemons.append(daemon)
 111
 112         return sorted(daemons, key=lambda k: k['id'])
 113
 114     def get(self, svc_id):
 115         # type: (str) -> dict
 116         daemon = {
 117             'rgw_metadata': [],
 118             'rgw_id': svc_id,
 119             'rgw_status': []
 120         }
 121         service = CephService.get_service('rgw', svc_id)
 122         if not service:
 123             raise cherrypy.NotFound('Service rgw {} is not available'.format(svc_id))
 124
 125         metadata = service['metadata']
 126         status = service['status']
 127         if 'json' in status:
 128             try:
 129                 status = json.loads(status['json'])
 130             except ValueError:
 131                 logger.warning('%s had invalid status json', service['id'])
 132                 status = {}
 133         else:
 134             logger.warning('%s has no key "json" in status', service['id'])
 135
 136         daemon['rgw_metadata'] = metadata
 137         daemon['rgw_status'] = status
 138         return daemon
 139
 140
 141 class RgwRESTController(RESTController):
 142     def proxy(self, daemon_name, method, path, params=None, json_response=True):
 143         try:
 144             instance = RgwClient.admin_instance(daemon_name=daemon_name)
 145             result = instance.proxy(method, path, params, None)
 146             if json_response:
 147                 result = json_str_to_object(result)
 148             return result
 149         except (DashboardException, RequestException) as e:
 150             http_status_code = e.status if isinstance(e, DashboardException) else 500
 151             raise DashboardException(e, http_status_code=http_status_code, component='rgw')
 152
 153
 154 @APIRouter('/rgw/site', Scope.RGW)
 155 @APIDoc("RGW Site Management API", "RgwSite")
 156 class RgwSite(RgwRESTController):
 157     def list(self, query=None, daemon_name=None):
 158         if query == 'placement-targets':
 159             return RgwClient.admin_instance(daemon_name=daemon_name).get_placement_targets()
 160         if query == 'realms':
 161             return RgwClient.admin_instance(daemon_name=daemon_name).get_realms()
 162         if query == 'default-realm':
 163             return RgwClient.admin_instance(daemon_name=daemon_name).get_default_realm()
 164
 165         # @TODO: for multisite: by default, retrieve cluster topology/map.
 166         raise DashboardException(http_status_code=501, component='rgw', msg='Not Implemented')
 167
 168
 169 @APIRouter('/rgw/bucket', Scope.RGW)
 170 @APIDoc("RGW Bucket Management API", "RgwBucket")
 171 class RgwBucket(RgwRESTController):
 172     def _append_bid(self, bucket):
 173         """
 174         Append the bucket identifier that looks like [<tenant>/]<bucket>.
 175         See http://docs.ceph.com/docs/nautilus/radosgw/multitenancy/ for
 176         more information.
 177         :param bucket: The bucket parameters.
 178         :type bucket: dict
 179         :return: The modified bucket parameters including the 'bid' parameter.
 180         :rtype: dict
 181         """
 182         if isinstance(bucket, dict):
 183             bucket['bid'] = '{}/{}'.format(bucket['tenant'], bucket['bucket']) \
 184                 if bucket['tenant'] else bucket['bucket']
 185         return bucket
 186
 187     def _get_versioning(self, owner, daemon_name, bucket_name):
 188         rgw_client = RgwClient.instance(owner, daemon_name)
 189         return rgw_client.get_bucket_versioning(bucket_name)
 190
 191     def _set_versioning(self, owner, daemon_name, bucket_name, versioning_state, mfa_delete,
 192                         mfa_token_serial, mfa_token_pin):
 193         bucket_versioning = self._get_versioning(owner, daemon_name, bucket_name)
 194         if versioning_state != bucket_versioning['Status']\
 195                 or (mfa_delete and mfa_delete != bucket_versioning['MfaDelete']):
 196             rgw_client = RgwClient.instance(owner, daemon_name)
 197             rgw_client.set_bucket_versioning(bucket_name, versioning_state, mfa_delete,
 198                                              mfa_token_serial, mfa_token_pin)
 199
 200     def _get_locking(self, owner, daemon_name, bucket_name):
 201         rgw_client = RgwClient.instance(owner, daemon_name)
 202         return rgw_client.get_bucket_locking(bucket_name)
 203
 204     def _set_locking(self, owner, daemon_name, bucket_name, mode,
 205                      retention_period_days, retention_period_years):
 206         rgw_client = RgwClient.instance(owner, daemon_name)
 207         return rgw_client.set_bucket_locking(bucket_name, mode,
 208                                              retention_period_days,
 209                                              retention_period_years)
 210
 211     @staticmethod
 212     def strip_tenant_from_bucket_name(bucket_name):
 213         # type (str) -> str
 214         """
 215         >>> RgwBucket.strip_tenant_from_bucket_name('tenant/bucket-name')
 216         'bucket-name'
 217         >>> RgwBucket.strip_tenant_from_bucket_name('bucket-name')
 218         'bucket-name'
 219         """
 220         return bucket_name[bucket_name.find('/') + 1:]
 221
 222     @staticmethod
 223     def get_s3_bucket_name(bucket_name, tenant=None):
 224         # type (str, str) -> str
 225         """
 226         >>> RgwBucket.get_s3_bucket_name('bucket-name', 'tenant')
 227         'tenant:bucket-name'
 228         >>> RgwBucket.get_s3_bucket_name('tenant/bucket-name', 'tenant')
 229         'tenant:bucket-name'
 230         >>> RgwBucket.get_s3_bucket_name('bucket-name')
 231         'bucket-name'
 232         """
 233         bucket_name = RgwBucket.strip_tenant_from_bucket_name(bucket_name)
 234         if tenant:
 235             bucket_name = '{}:{}'.format(tenant, bucket_name)
 236         return bucket_name
 237
 238     @RESTController.MethodMap(version=APIVersion(1, 1))  # type: ignore
 239     def list(self, stats: bool = False, daemon_name: Optional[str] = None,
 240              uid: Optional[str] = None) -> List[Union[str, Dict[str, Any]]]:
 241         query_params = f'?stats={str_to_bool(stats)}'
 242         if uid and uid.strip():
 243             query_params = f'{query_params}&uid={uid.strip()}'
 244         result = self.proxy(daemon_name, 'GET', 'bucket{}'.format(query_params))
 245
 246         if stats:
 247             result = [self._append_bid(bucket) for bucket in result]
 248
 249         return result
 250
 251     def get(self, bucket, daemon_name=None):
 252         # type: (str, Optional[str]) -> dict
 253         result = self.proxy(daemon_name, 'GET', 'bucket', {'bucket': bucket})
 254         bucket_name = RgwBucket.get_s3_bucket_name(result['bucket'],
 255                                                    result['tenant'])
 256
 257         # Append the versioning configuration.
 258         versioning = self._get_versioning(result['owner'], daemon_name, bucket_name)
 259         result['versioning'] = versioning['Status']
 260         result['mfa_delete'] = versioning['MfaDelete']
 261
 262         # Append the locking configuration.
 263         locking = self._get_locking(result['owner'], daemon_name, bucket_name)
 264         result.update(locking)
 265
 266         return self._append_bid(result)
 267
 268     @allow_empty_body
 269     def create(self, bucket, uid, zonegroup=None, placement_target=None,
 270                lock_enabled='false', lock_mode=None,
 271                lock_retention_period_days=None,
 272                lock_retention_period_years=None, daemon_name=None):
 273         lock_enabled = str_to_bool(lock_enabled)
 274         try:
 275             rgw_client = RgwClient.instance(uid, daemon_name)
 276             result = rgw_client.create_bucket(bucket, zonegroup,
 277                                               placement_target,
 278                                               lock_enabled)
 279             if lock_enabled:
 280                 self._set_locking(uid, daemon_name, bucket, lock_mode,
 281                                   lock_retention_period_days,
 282                                   lock_retention_period_years)
 283             return result
 284         except RequestException as e:  # pragma: no cover - handling is too obvious
 285             raise DashboardException(e, http_status_code=500, component='rgw')
 286
 287     @allow_empty_body
 288     def set(self, bucket, bucket_id, uid, versioning_state=None,
 289             mfa_delete=None, mfa_token_serial=None, mfa_token_pin=None,
 290             lock_mode=None, lock_retention_period_days=None,
 291             lock_retention_period_years=None, daemon_name=None):
 292         # When linking a non-tenant-user owned bucket to a tenanted user, we
 293         # need to prefix bucket name with '/'. e.g. photos -> /photos
 294         if '$' in uid and '/' not in bucket:
 295             bucket = '/{}'.format(bucket)
 296
 297         # Link bucket to new user:
 298         result = self.proxy(daemon_name,
 299                             'PUT',
 300                             'bucket', {
 301                                 'bucket': bucket,
 302                                 'bucket-id': bucket_id,
 303                                 'uid': uid
 304                             },
 305                             json_response=False)
 306
 307         uid_tenant = uid[:uid.find('$')] if uid.find('$') >= 0 else None
 308         bucket_name = RgwBucket.get_s3_bucket_name(bucket, uid_tenant)
 309
 310         locking = self._get_locking(uid, daemon_name, bucket_name)
 311         if versioning_state:
 312             if versioning_state == 'Suspended' and locking['lock_enabled']:
 313                 raise DashboardException(msg='Bucket versioning cannot be disabled/suspended '
 314                                              'on buckets with object lock enabled ',
 315                                              http_status_code=409, component='rgw')
 316             self._set_versioning(uid, daemon_name, bucket_name, versioning_state,
 317                                  mfa_delete, mfa_token_serial, mfa_token_pin)
 318
 319         # Update locking if it is enabled.
 320         if locking['lock_enabled']:
 321             self._set_locking(uid, daemon_name, bucket_name, lock_mode,
 322                               lock_retention_period_days,
 323                               lock_retention_period_years)
 324
 325         return self._append_bid(result)
 326
 327     def delete(self, bucket, purge_objects='true', daemon_name=None):
 328         return self.proxy(daemon_name, 'DELETE', 'bucket', {
 329             'bucket': bucket,
 330             'purge-objects': purge_objects
 331         }, json_response=False)
 332
 333
 334 @APIRouter('/rgw/user', Scope.RGW)
 335 @APIDoc("RGW User Management API", "RgwUser")
 336 class RgwUser(RgwRESTController):
 337     def _append_uid(self, user):
 338         """
 339         Append the user identifier that looks like [<tenant>$]<user>.
 340         See http://docs.ceph.com/docs/jewel/radosgw/multitenancy/ for
 341         more information.
 342         :param user: The user parameters.
 343         :type user: dict
 344         :return: The modified user parameters including the 'uid' parameter.
 345         :rtype: dict
 346         """
 347         if isinstance(user, dict):
 348             user['uid'] = '{}${}'.format(user['tenant'], user['user_id']) \
 349                 if user['tenant'] else user['user_id']
 350         return user
 351
 352     @staticmethod
 353     def _keys_allowed():
 354         permissions = AuthManager.get_user(JwtManager.get_username()).permissions_dict()
 355         edit_permissions = [Permission.CREATE, Permission.UPDATE, Permission.DELETE]
 356         return Scope.RGW in permissions and Permission.READ in permissions[Scope.RGW] \
 357             and len(set(edit_permissions).intersection(set(permissions[Scope.RGW]))) > 0
 358
 359     @EndpointDoc("Display RGW Users",
 360                  responses={200: RGW_USER_SCHEMA})
 361     def list(self, daemon_name=None):
 362         # type: (Optional[str]) -> List[str]
 363         users = []  # type: List[str]
 364         marker = None
 365         while True:
 366             params = {}  # type: dict
 367             if marker:
 368                 params['marker'] = marker
 369             result = self.proxy(daemon_name, 'GET', 'user?list', params)
 370             users.extend(result['keys'])
 371             if not result['truncated']:
 372                 break
 373             # Make sure there is a marker.
 374             assert result['marker']
 375             # Make sure the marker has changed.
 376             assert marker != result['marker']
 377             marker = result['marker']
 378         return users
 379
 380     def get(self, uid, daemon_name=None, stats=True) -> dict:
 381         query_params = '?stats' if stats else ''
 382         result = self.proxy(daemon_name, 'GET', 'user{}'.format(query_params),
 383                             {'uid': uid, 'stats': stats})
 384         if not self._keys_allowed():
 385             del result['keys']
 386             del result['swift_keys']
 387         return self._append_uid(result)
 388
 389     @Endpoint()
 390     @ReadPermission
 391     def get_emails(self, daemon_name=None):
 392         # type: (Optional[str]) -> List[str]
 393         emails = []
 394         for uid in json.loads(self.list(daemon_name)):  # type: ignore
 395             user = json.loads(self.get(uid, daemon_name))  # type: ignore
 396             if user["email"]:
 397                 emails.append(user["email"])
 398         return emails
 399
 400     @allow_empty_body
 401     def create(self, uid, display_name, email=None, max_buckets=None,
 402                suspended=None, generate_key=None, access_key=None,
 403                secret_key=None, daemon_name=None):
 404         params = {'uid': uid}
 405         if display_name is not None:
 406             params['display-name'] = display_name
 407         if email is not None:
 408             params['email'] = email
 409         if max_buckets is not None:
 410             params['max-buckets'] = max_buckets
 411         if suspended is not None:
 412             params['suspended'] = suspended
 413         if generate_key is not None:
 414             params['generate-key'] = generate_key
 415         if access_key is not None:
 416             params['access-key'] = access_key
 417         if secret_key is not None:
 418             params['secret-key'] = secret_key
 419         result = self.proxy(daemon_name, 'PUT', 'user', params)
 420         return self._append_uid(result)
 421
 422     @allow_empty_body
 423     def set(self, uid, display_name=None, email=None, max_buckets=None,
 424             suspended=None, daemon_name=None):
 425         params = {'uid': uid}
 426         if display_name is not None:
 427             params['display-name'] = display_name
 428         if email is not None:
 429             params['email'] = email
 430         if max_buckets is not None:
 431             params['max-buckets'] = max_buckets
 432         if suspended is not None:
 433             params['suspended'] = suspended
 434         result = self.proxy(daemon_name, 'POST', 'user', params)
 435         return self._append_uid(result)
 436
 437     def delete(self, uid, daemon_name=None):
 438         try:
 439             instance = RgwClient.admin_instance(daemon_name=daemon_name)
 440             # Ensure the user is not configured to access the RGW Object Gateway.
 441             if instance.userid == uid:
 442                 raise DashboardException(msg='Unable to delete "{}" - this user '
 443                                              'account is required for managing the '
 444                                              'Object Gateway'.format(uid))
 445             # Finally redirect request to the RGW proxy.
 446             return self.proxy(daemon_name, 'DELETE', 'user', {'uid': uid}, json_response=False)
 447         except (DashboardException, RequestException) as e:  # pragma: no cover
 448             raise DashboardException(e, component='rgw')
 449
 450     # pylint: disable=redefined-builtin
 451     @RESTController.Resource(method='POST', path='/capability', status=201)
 452     @allow_empty_body
 453     def create_cap(self, uid, type, perm, daemon_name=None):
 454         return self.proxy(daemon_name, 'PUT', 'user?caps', {
 455             'uid': uid,
 456             'user-caps': '{}={}'.format(type, perm)
 457         })
 458
 459     # pylint: disable=redefined-builtin
 460     @RESTController.Resource(method='DELETE', path='/capability', status=204)
 461     def delete_cap(self, uid, type, perm, daemon_name=None):
 462         return self.proxy(daemon_name, 'DELETE', 'user?caps', {
 463             'uid': uid,
 464             'user-caps': '{}={}'.format(type, perm)
 465         })
 466
 467     @RESTController.Resource(method='POST', path='/key', status=201)
 468     @allow_empty_body
 469     def create_key(self, uid, key_type='s3', subuser=None, generate_key='true',
 470                    access_key=None, secret_key=None, daemon_name=None):
 471         params = {'uid': uid, 'key-type': key_type, 'generate-key': generate_key}
 472         if subuser is not None:
 473             params['subuser'] = subuser
 474         if access_key is not None:
 475             params['access-key'] = access_key
 476         if secret_key is not None:
 477             params['secret-key'] = secret_key
 478         return self.proxy(daemon_name, 'PUT', 'user?key', params)
 479
 480     @RESTController.Resource(method='DELETE', path='/key', status=204)
 481     def delete_key(self, uid, key_type='s3', subuser=None, access_key=None, daemon_name=None):
 482         params = {'uid': uid, 'key-type': key_type}
 483         if subuser is not None:
 484             params['subuser'] = subuser
 485         if access_key is not None:
 486             params['access-key'] = access_key
 487         return self.proxy(daemon_name, 'DELETE', 'user?key', params, json_response=False)
 488
 489     @RESTController.Resource(method='GET', path='/quota')
 490     def get_quota(self, uid, daemon_name=None):
 491         return self.proxy(daemon_name, 'GET', 'user?quota', {'uid': uid})
 492
 493     @RESTController.Resource(method='PUT', path='/quota')
 494     @allow_empty_body
 495     def set_quota(self, uid, quota_type, enabled, max_size_kb, max_objects, daemon_name=None):
 496         return self.proxy(daemon_name, 'PUT', 'user?quota', {
 497             'uid': uid,
 498             'quota-type': quota_type,
 499             'enabled': enabled,
 500             'max-size-kb': max_size_kb,
 501             'max-objects': max_objects
 502         }, json_response=False)
 503
 504     @RESTController.Resource(method='POST', path='/subuser', status=201)
 505     @allow_empty_body
 506     def create_subuser(self, uid, subuser, access, key_type='s3',
 507                        generate_secret='true', access_key=None,
 508                        secret_key=None, daemon_name=None):
 509         return self.proxy(daemon_name, 'PUT', 'user', {
 510             'uid': uid,
 511             'subuser': subuser,
 512             'key-type': key_type,
 513             'access': access,
 514             'generate-secret': generate_secret,
 515             'access-key': access_key,
 516             'secret-key': secret_key
 517         })
 518
 519     @RESTController.Resource(method='DELETE', path='/subuser/{subuser}', status=204)
 520     def delete_subuser(self, uid, subuser, purge_keys='true', daemon_name=None):
 521         """
 522         :param purge_keys: Set to False to do not purge the keys.
 523                            Note, this only works for s3 subusers.
 524         """
 525         return self.proxy(daemon_name, 'DELETE', 'user', {
 526             'uid': uid,
 527             'subuser': subuser,
 528             'purge-keys': purge_keys
 529         }, json_response=False)