ceph/src/pybind/mgr/dashboard/services/rgw_client.py

   1 # -*- coding: utf-8 -*-
   2 from __future__ import absolute_import
   3
   4 import re
   5 import logging
   6 import ipaddress
   7 from distutils.util import strtobool
   8 import xml.etree.ElementTree as ET  # noqa: N814
   9 import six
  10 from ..awsauth import S3Auth
  11 from ..exceptions import DashboardException
  12 from ..settings import Settings, Options
  13 from ..rest_client import RestClient, RequestException
  14 from ..tools import build_url, dict_contains_path, json_str_to_object,\
  15                     partial_dict, dict_get
  16 from .. import mgr
  17
  18 try:
  19     from typing import Dict, List, Optional  # pylint: disable=unused-import
  20 except ImportError:
  21     pass  # For typing only
  22
  23 logger = logging.getLogger('rgw_client')
  24
  25
  26 class NoCredentialsException(RequestException):
  27     def __init__(self):
  28         super(NoCredentialsException, self).__init__(
  29             'No RGW credentials found, '
  30             'please consult the documentation on how to enable RGW for '
  31             'the dashboard.')
  32
  33
  34 def _determine_rgw_addr():
  35     """
  36     Get a RGW daemon to determine the configured host (IP address) and port.
  37     Note, the service id of the RGW daemons may differ depending on the setup.
  38     Example 1:
  39     {
  40         ...
  41         'services': {
  42             'rgw': {
  43                 'daemons': {
  44                     'summary': '',
  45                     '0': {
  46                         ...
  47                         'addr': '[2001:db8:85a3::8a2e:370:7334]:49774/1534999298',
  48                         'metadata': {
  49                             'frontend_config#0': 'civetweb port=7280',
  50                         }
  51                         ...
  52                     }
  53                 }
  54             }
  55         }
  56     }
  57     Example 2:
  58     {
  59         ...
  60         'services': {
  61             'rgw': {
  62                 'daemons': {
  63                     'summary': '',
  64                     'rgw': {
  65                         ...
  66                         'addr': '192.168.178.3:49774/1534999298',
  67                         'metadata': {
  68                             'frontend_config#0': 'civetweb port=8000',
  69                         }
  70                         ...
  71                     }
  72                 }
  73             }
  74         }
  75     }
  76     """
  77     service_map = mgr.get('service_map')
  78     if not dict_contains_path(service_map, ['services', 'rgw', 'daemons']):
  79         raise LookupError('No RGW found')
  80     daemon = None
  81     daemons = service_map['services']['rgw']['daemons']
  82     for key in daemons.keys():
  83         if dict_contains_path(daemons[key], ['metadata', 'frontend_config#0']):
  84             daemon = daemons[key]
  85             break
  86     if daemon is None:
  87         raise LookupError('No RGW daemon found')
  88
  89     addr = _parse_addr(daemon['addr'])
  90     port, ssl = _parse_frontend_config(daemon['metadata']['frontend_config#0'])
  91
  92     return addr, port, ssl
  93
  94
  95 def _parse_addr(value):
  96     """
  97     Get the IP address the RGW is running on.
  98
  99     >>> _parse_addr('192.168.178.3:49774/1534999298')
 100     '192.168.178.3'
 101
 102     >>> _parse_addr('[2001:db8:85a3::8a2e:370:7334]:49774/1534999298')
 103     '2001:db8:85a3::8a2e:370:7334'
 104
 105     >>> _parse_addr('xyz')
 106     Traceback (most recent call last):
 107     ...
 108     LookupError: Failed to determine RGW address
 109
 110     >>> _parse_addr('192.168.178.a:8080/123456789')
 111     Traceback (most recent call last):
 112     ...
 113     LookupError: Invalid RGW address '192.168.178.a' found
 114
 115     >>> _parse_addr('[2001:0db8:1234]:443/123456789')
 116     Traceback (most recent call last):
 117     ...
 118     LookupError: Invalid RGW address '2001:0db8:1234' found
 119
 120     >>> _parse_addr('2001:0db8::1234:49774/1534999298')
 121     Traceback (most recent call last):
 122     ...
 123     LookupError: Failed to determine RGW address
 124
 125     :param value: The string to process. The syntax is '<HOST>:<PORT>/<NONCE>'.
 126     :type: str
 127     :raises LookupError if parsing fails to determine the IP address.
 128     :return: The IP address.
 129     :rtype: str
 130     """
 131     match = re.search(r'^(\[)?(?(1)([^\]]+)\]|([^:]+)):\d+/\d+?', value)
 132     if match:
 133         # IPv4:
 134         #   Group 0: 192.168.178.3:49774/1534999298
 135         #   Group 3: 192.168.178.3
 136         # IPv6:
 137         #   Group 0: [2001:db8:85a3::8a2e:370:7334]:49774/1534999298
 138         #   Group 1: [
 139         #   Group 2: 2001:db8:85a3::8a2e:370:7334
 140         addr = match.group(3) if match.group(3) else match.group(2)
 141         try:
 142             ipaddress.ip_address(six.u(addr))
 143             return addr
 144         except ValueError:
 145             raise LookupError('Invalid RGW address \'{}\' found'.format(addr))
 146     raise LookupError('Failed to determine RGW address')
 147
 148
 149 def _parse_frontend_config(config):
 150     """
 151     Get the port the RGW is running on. Due the complexity of the
 152     syntax not all variations are supported.
 153
 154     If there are multiple (ssl_)ports/(ssl_)endpoints options, then
 155     the first found option will be returned.
 156
 157     Get more details about the configuration syntax here:
 158     http://docs.ceph.com/docs/master/radosgw/frontends/
 159     https://civetweb.github.io/civetweb/UserManual.html
 160
 161     :param config: The configuration string to parse.
 162     :type config: str
 163     :raises LookupError if parsing fails to determine the port.
 164     :return: A tuple containing the port number and the information
 165              whether SSL is used.
 166     :rtype: (int, boolean)
 167     """
 168     match = re.search(r'^(beast|civetweb)\s+.+$', config)
 169     if match:
 170         if match.group(1) == 'beast':
 171             match = re.search(r'(port|ssl_port|endpoint|ssl_endpoint)=(.+)',
 172                               config)
 173             if match:
 174                 option_name = match.group(1)
 175                 if option_name in ['port', 'ssl_port']:
 176                     match = re.search(r'(\d+)', match.group(2))
 177                     if match:
 178                         port = int(match.group(1))
 179                         ssl = option_name == 'ssl_port'
 180                         return port, ssl
 181                 if option_name in ['endpoint', 'ssl_endpoint']:
 182                     match = re.search(r'([\d.]+|\[.+\])(:(\d+))?',
 183                                       match.group(2))  # type: ignore
 184                     if match:
 185                         port = int(match.group(3)) if \
 186                             match.group(2) is not None else 443 if \
 187                             option_name == 'ssl_endpoint' else \
 188                             80
 189                         ssl = option_name == 'ssl_endpoint'
 190                         return port, ssl
 191         if match.group(1) == 'civetweb':  # type: ignore
 192             match = re.search(r'port=(.*:)?(\d+)(s)?', config)
 193             if match:
 194                 port = int(match.group(2))
 195                 ssl = match.group(3) == 's'
 196                 return port, ssl
 197     raise LookupError('Failed to determine RGW port from "{}"'.format(config))
 198
 199
 200 class RgwClient(RestClient):
 201     _SYSTEM_USERID = None
 202     _ADMIN_PATH = None
 203     _host = None
 204     _port = None
 205     _ssl = None
 206     _user_instances = {}  # type: Dict[str, RgwClient]
 207     _rgw_settings_snapshot = None
 208
 209     @staticmethod
 210     def _load_settings():
 211         # The API access key and secret key are mandatory for a minimal configuration.
 212         if not (Settings.RGW_API_ACCESS_KEY and Settings.RGW_API_SECRET_KEY):
 213             logger.warning('No credentials found, please consult the '
 214                            'documentation about how to enable RGW for the '
 215                            'dashboard.')
 216             raise NoCredentialsException()
 217
 218         if Options.has_default_value('RGW_API_HOST') and \
 219                 Options.has_default_value('RGW_API_PORT') and \
 220                 Options.has_default_value('RGW_API_SCHEME'):
 221             host, port, ssl = _determine_rgw_addr()
 222         else:
 223             host = Settings.RGW_API_HOST
 224             port = Settings.RGW_API_PORT
 225             ssl = Settings.RGW_API_SCHEME == 'https'
 226
 227         RgwClient._host = host
 228         RgwClient._port = port
 229         RgwClient._ssl = ssl
 230         RgwClient._ADMIN_PATH = Settings.RGW_API_ADMIN_RESOURCE
 231
 232         # Create an instance using the configured settings.
 233         instance = RgwClient(Settings.RGW_API_USER_ID,
 234                              Settings.RGW_API_ACCESS_KEY,
 235                              Settings.RGW_API_SECRET_KEY)
 236
 237         RgwClient._SYSTEM_USERID = instance.userid
 238
 239         # Append the instance to the internal map.
 240         RgwClient._user_instances[RgwClient._SYSTEM_USERID] = instance
 241
 242     def _get_daemon_zone_info(self):  # type: () -> dict
 243         return json_str_to_object(self.proxy('GET', 'config?type=zone', None, None))
 244
 245     def _get_daemon_zonegroup_map(self):  # type: () -> List[dict]
 246         zonegroups = json_str_to_object(
 247             self.proxy('GET', 'config?type=zonegroup-map', None, None)
 248         )
 249
 250         return [partial_dict(
 251             zonegroup['val'],
 252             ['api_name', 'zones']
 253         ) for zonegroup in zonegroups['zonegroups']]
 254
 255     @staticmethod
 256     def _rgw_settings():
 257         return (Settings.RGW_API_HOST,
 258                 Settings.RGW_API_PORT,
 259                 Settings.RGW_API_ACCESS_KEY,
 260                 Settings.RGW_API_SECRET_KEY,
 261                 Settings.RGW_API_ADMIN_RESOURCE,
 262                 Settings.RGW_API_SCHEME,
 263                 Settings.RGW_API_USER_ID,
 264                 Settings.RGW_API_SSL_VERIFY)
 265
 266     @staticmethod
 267     def instance(userid):
 268         # type: (Optional[str]) -> RgwClient
 269         # Discard all cached instances if any rgw setting has changed
 270         if RgwClient._rgw_settings_snapshot != RgwClient._rgw_settings():
 271             RgwClient._rgw_settings_snapshot = RgwClient._rgw_settings()
 272             RgwClient._user_instances.clear()
 273
 274         if not RgwClient._user_instances:
 275             RgwClient._load_settings()
 276
 277         if not userid:
 278             userid = RgwClient._SYSTEM_USERID
 279
 280         if userid not in RgwClient._user_instances:
 281             # Get the access and secret keys for the specified user.
 282             keys = RgwClient.admin_instance().get_user_keys(userid)
 283             if not keys:
 284                 raise RequestException(
 285                     "User '{}' does not have any keys configured.".format(
 286                         userid))
 287
 288             # Create an instance and append it to the internal map.
 289             RgwClient._user_instances[userid] = RgwClient(userid,  # type: ignore
 290                                                           keys['access_key'],
 291                                                           keys['secret_key'])
 292
 293         return RgwClient._user_instances[userid]  # type: ignore
 294
 295     @staticmethod
 296     def admin_instance():
 297         return RgwClient.instance(RgwClient._SYSTEM_USERID)
 298
 299     def _reset_login(self):
 300         if self.userid != RgwClient._SYSTEM_USERID:
 301             logger.info("Fetching new keys for user: %s", self.userid)
 302             keys = RgwClient.admin_instance().get_user_keys(self.userid)
 303             self.auth = S3Auth(keys['access_key'], keys['secret_key'],
 304                                service_url=self.service_url)
 305         else:
 306             raise RequestException('Authentication failed for the "{}" user: wrong credentials'
 307                                    .format(self.userid), status_code=401)
 308
 309     def __init__(self,  # pylint: disable-msg=R0913
 310                  userid,
 311                  access_key,
 312                  secret_key,
 313                  host=None,
 314                  port=None,
 315                  admin_path=None,
 316                  ssl=False):
 317
 318         if not host and not RgwClient._host:
 319             RgwClient._load_settings()
 320         host = host if host else RgwClient._host
 321         port = port if port else RgwClient._port
 322         admin_path = admin_path if admin_path else RgwClient._ADMIN_PATH
 323         ssl = ssl if ssl else RgwClient._ssl
 324         ssl_verify = Settings.RGW_API_SSL_VERIFY
 325
 326         self.service_url = build_url(host=host, port=port)
 327         self.admin_path = admin_path
 328
 329         s3auth = S3Auth(access_key, secret_key, service_url=self.service_url)
 330         super(RgwClient, self).__init__(host, port, 'RGW', ssl, s3auth, ssl_verify=ssl_verify)
 331
 332         # If user ID is not set, then try to get it via the RGW Admin Ops API.
 333         self.userid = userid if userid else self._get_user_id(self.admin_path)  # type: str
 334
 335         logger.info("Created new connection: user=%s, host=%s, port=%s, ssl=%d, sslverify=%d",
 336                     self.userid, host, port, ssl, ssl_verify)
 337
 338     @RestClient.api_get('/', resp_structure='[0] > (ID & DisplayName)')
 339     def is_service_online(self, request=None):
 340         """
 341         Consider the service as online if the response contains the
 342         specified keys. Nothing more is checked here.
 343         """
 344         _ = request({'format': 'json'})
 345         return True
 346
 347     @RestClient.api_get('/{admin_path}/metadata/user?myself',
 348                         resp_structure='data > user_id')
 349     def _get_user_id(self, admin_path, request=None):
 350         # pylint: disable=unused-argument
 351         """
 352         Get the user ID of the user that is used to communicate with the
 353         RGW Admin Ops API.
 354         :rtype: str
 355         :return: The user ID of the user that is used to sign the
 356                  RGW Admin Ops API calls.
 357         """
 358         response = request()
 359         return response['data']['user_id']
 360
 361     @RestClient.api_get('/{admin_path}/metadata/user', resp_structure='[+]')
 362     def _user_exists(self, admin_path, user_id, request=None):
 363         # pylint: disable=unused-argument
 364         response = request()
 365         if user_id:
 366             return user_id in response
 367         return self.userid in response
 368
 369     def user_exists(self, user_id=None):
 370         return self._user_exists(self.admin_path, user_id)
 371
 372     @RestClient.api_get('/{admin_path}/metadata/user?key={userid}',
 373                         resp_structure='data > system')
 374     def _is_system_user(self, admin_path, userid, request=None):
 375         # pylint: disable=unused-argument
 376         response = request()
 377         return strtobool(response['data']['system'])
 378
 379     def is_system_user(self):
 380         return self._is_system_user(self.admin_path, self.userid)
 381
 382     @RestClient.api_get(
 383         '/{admin_path}/user',
 384         resp_structure='tenant & user_id & email & keys[*] > '
 385         ' (user & access_key & secret_key)')
 386     def _admin_get_user_keys(self, admin_path, userid, request=None):
 387         # pylint: disable=unused-argument
 388         colon_idx = userid.find(':')
 389         user = userid if colon_idx == -1 else userid[:colon_idx]
 390         response = request({'uid': user})
 391         for key in response['keys']:
 392             if key['user'] == userid:
 393                 return {
 394                     'access_key': key['access_key'],
 395                     'secret_key': key['secret_key']
 396                 }
 397         return None
 398
 399     def get_user_keys(self, userid):
 400         return self._admin_get_user_keys(self.admin_path, userid)
 401
 402     @RestClient.api('/{admin_path}/{path}')
 403     def _proxy_request(
 404             self,  # pylint: disable=too-many-arguments
 405             admin_path,
 406             path,
 407             method,
 408             params,
 409             data,
 410             request=None):
 411         # pylint: disable=unused-argument
 412         return request(method=method, params=params, data=data,
 413                        raw_content=True)
 414
 415     def proxy(self, method, path, params, data):
 416         logger.debug("proxying method=%s path=%s params=%s data=%s",
 417                      method, path, params, data)
 418         return self._proxy_request(self.admin_path, path, method,
 419                                    params, data)
 420
 421     @RestClient.api_get('/', resp_structure='[1][*] > Name')
 422     def get_buckets(self, request=None):
 423         """
 424         Get a list of names from all existing buckets of this user.
 425         :return: Returns a list of bucket names.
 426         """
 427         response = request({'format': 'json'})
 428         return [bucket['Name'] for bucket in response[1]]
 429
 430     @RestClient.api_get('/{bucket_name}')
 431     def bucket_exists(self, bucket_name, userid, request=None):
 432         """
 433         Check if the specified bucket exists for this user.
 434         :param bucket_name: The name of the bucket.
 435         :return: Returns True if the bucket exists, otherwise False.
 436         """
 437         # pylint: disable=unused-argument
 438         try:
 439             request()
 440             my_buckets = self.get_buckets()
 441             if bucket_name not in my_buckets:
 442                 raise RequestException(
 443                     'Bucket "{}" belongs to other user'.format(bucket_name),
 444                     403)
 445             return True
 446         except RequestException as e:
 447             if e.status_code == 404:
 448                 return False
 449
 450             raise e
 451
 452     @RestClient.api_put('/{bucket_name}')
 453     def create_bucket(self, bucket_name, zonegroup=None,
 454                       placement_target=None, lock_enabled=False,
 455                       request=None):
 456         logger.info("Creating bucket: %s, zonegroup: %s, placement_target: %s",
 457                     bucket_name, zonegroup, placement_target)
 458         data = None
 459         if zonegroup and placement_target:
 460             create_bucket_configuration = ET.Element('CreateBucketConfiguration')
 461             location_constraint = ET.SubElement(create_bucket_configuration, 'LocationConstraint')
 462             location_constraint.text = '{}:{}'.format(zonegroup, placement_target)
 463             data = ET.tostring(create_bucket_configuration, encoding='unicode')
 464
 465         headers = None  # type: Optional[dict]
 466         if lock_enabled:
 467             headers = {'x-amz-bucket-object-lock-enabled': 'true'}
 468
 469         return request(data=data, headers=headers)
 470
 471     def get_placement_targets(self):  # type: () -> dict
 472         zone = self._get_daemon_zone_info()
 473         # A zone without realm id can only belong to default zonegroup.
 474         zonegroup_name = 'default'
 475         if zone['realm_id']:
 476             zonegroup_map = self._get_daemon_zonegroup_map()
 477             for zonegroup in zonegroup_map:
 478                 for realm_zone in zonegroup['zones']:
 479                     if realm_zone['id'] == zone['id']:
 480                         zonegroup_name = zonegroup['api_name']
 481                         break
 482
 483         placement_targets = []  # type: List[Dict]
 484         for placement_pool in zone['placement_pools']:
 485             placement_targets.append(
 486                 {
 487                     'name': placement_pool['key'],
 488                     'data_pool': placement_pool['val']['storage_classes']['STANDARD']['data_pool']
 489                 }
 490             )
 491
 492         return {'zonegroup': zonegroup_name, 'placement_targets': placement_targets}
 493
 494     @RestClient.api_get('/{bucket_name}?versioning')
 495     def get_bucket_versioning(self, bucket_name, request=None):
 496         """
 497         Get bucket versioning.
 498         :param str bucket_name: the name of the bucket.
 499         :return: versioning info
 500         :rtype: Dict
 501         """
 502         # pylint: disable=unused-argument
 503         result = request()
 504         if 'Status' not in result:
 505             result['Status'] = 'Suspended'
 506         if 'MfaDelete' not in result:
 507             result['MfaDelete'] = 'Disabled'
 508         return result
 509
 510     @RestClient.api_put('/{bucket_name}?versioning')
 511     def set_bucket_versioning(self, bucket_name, versioning_state, mfa_delete,
 512                               mfa_token_serial, mfa_token_pin, request=None):
 513         """
 514         Set bucket versioning.
 515         :param str bucket_name: the name of the bucket.
 516         :param str versioning_state:
 517             https://docs.aws.amazon.com/AmazonS3/latest/API/RESTBucketPUTVersioningStatus.html
 518         :param str mfa_delete: MFA Delete state.
 519         :param str mfa_token_serial:
 520             https://docs.ceph.com/docs/master/radosgw/mfa/
 521         :param str mfa_token_pin: value of a TOTP token at a certain time (auth code)
 522         :return: None
 523         """
 524         # pylint: disable=unused-argument
 525         versioning_configuration = ET.Element('VersioningConfiguration')
 526         status_element = ET.SubElement(versioning_configuration, 'Status')
 527         status_element.text = versioning_state
 528
 529         headers = {}
 530         if mfa_delete and mfa_token_serial and mfa_token_pin:
 531             headers['x-amz-mfa'] = '{} {}'.format(mfa_token_serial, mfa_token_pin)
 532             mfa_delete_element = ET.SubElement(versioning_configuration, 'MfaDelete')
 533             mfa_delete_element.text = mfa_delete
 534
 535         data = ET.tostring(versioning_configuration, encoding='unicode')
 536
 537         try:
 538             request(data=data, headers=headers)
 539         except RequestException as error:
 540             msg = str(error)
 541             if error.status_code == 403:
 542                 msg = 'Bad MFA credentials: {}'.format(msg)
 543             # Avoid dashboard GUI redirections caused by status code (403, ...):
 544             http_status_code = 400 if 400 <= error.status_code < 500 else error.status_code
 545             raise DashboardException(msg=msg,
 546                                      http_status_code=http_status_code,
 547                                      component='rgw')
 548
 549     @RestClient.api_get('/{bucket_name}?object-lock')
 550     def get_bucket_locking(self, bucket_name, request=None):
 551         # type: (str, Optional[object]) -> dict
 552         """
 553         Gets the locking configuration for a bucket. The locking
 554         configuration will be applied by default to every new object
 555         placed in the specified bucket.
 556         :param bucket_name: The name of the bucket.
 557         :type bucket_name: str
 558         :return: The locking configuration.
 559         :rtype: Dict
 560         """
 561         # pylint: disable=unused-argument
 562
 563         # Try to get the Object Lock configuration. If there is none,
 564         # then return default values.
 565         try:
 566             result = request()  # type: ignore
 567             return {
 568                 'lock_enabled': dict_get(result, 'ObjectLockEnabled') == 'Enabled',
 569                 'lock_mode': dict_get(result, 'Rule.DefaultRetention.Mode'),
 570                 'lock_retention_period_days': dict_get(result, 'Rule.DefaultRetention.Days', 0),
 571                 'lock_retention_period_years': dict_get(result, 'Rule.DefaultRetention.Years', 0)
 572             }
 573         except RequestException as e:
 574             if e.content:
 575                 content = json_str_to_object(e.content)
 576                 if content.get(
 577                         'Code') == 'ObjectLockConfigurationNotFoundError':
 578                     return {
 579                         'lock_enabled': False,
 580                         'lock_mode': 'compliance',
 581                         'lock_retention_period_days': None,
 582                         'lock_retention_period_years': None
 583                     }
 584             raise e
 585
 586     @RestClient.api_put('/{bucket_name}?object-lock')
 587     def set_bucket_locking(self,
 588                            bucket_name,
 589                            mode,
 590                            retention_period_days,
 591                            retention_period_years,
 592                            request=None):
 593         # type: (str, str, int, int, Optional[object]) -> None
 594         """
 595         Places the locking configuration on the specified bucket. The
 596         locking configuration will be applied by default to every new
 597         object placed in the specified bucket.
 598         :param bucket_name: The name of the bucket.
 599         :type bucket_name: str
 600         :param mode: The lock mode, e.g. `COMPLIANCE` or `GOVERNANCE`.
 601         :type mode: str
 602         :param retention_period_days:
 603         :type retention_period_days: int
 604         :param retention_period_years:
 605         :type retention_period_years: int
 606         :rtype: None
 607         """
 608         # pylint: disable=unused-argument
 609
 610         # Do some validations.
 611         if retention_period_days and retention_period_years:
 612             # https://docs.aws.amazon.com/AmazonS3/latest/API/archive-RESTBucketPUTObjectLockConfiguration.html
 613             msg = "Retention period requires either Days or Years. "\
 614                 "You can't specify both at the same time."
 615             raise DashboardException(msg=msg, component='rgw')
 616         if not retention_period_days and not retention_period_years:
 617             msg = "Retention period requires either Days or Years. "\
 618                 "You must specify at least one."
 619             raise DashboardException(msg=msg, component='rgw')
 620
 621         # Generate the XML data like this:
 622         # <ObjectLockConfiguration>
 623         #    <ObjectLockEnabled>string</ObjectLockEnabled>
 624         #    <Rule>
 625         #       <DefaultRetention>
 626         #          <Days>integer</Days>
 627         #          <Mode>string</Mode>
 628         #          <Years>integer</Years>
 629         #       </DefaultRetention>
 630         #    </Rule>
 631         # </ObjectLockConfiguration>
 632         locking_configuration = ET.Element('ObjectLockConfiguration')
 633         enabled_element = ET.SubElement(locking_configuration,
 634                                         'ObjectLockEnabled')
 635         enabled_element.text = 'Enabled'  # Locking can't be disabled.
 636         rule_element = ET.SubElement(locking_configuration, 'Rule')
 637         default_retention_element = ET.SubElement(rule_element,
 638                                                   'DefaultRetention')
 639         mode_element = ET.SubElement(default_retention_element, 'Mode')
 640         mode_element.text = mode.upper()
 641         if retention_period_days:
 642             days_element = ET.SubElement(default_retention_element, 'Days')
 643             days_element.text = str(retention_period_days)
 644         if retention_period_years:
 645             years_element = ET.SubElement(default_retention_element, 'Years')
 646             years_element.text = str(retention_period_years)
 647
 648         data = ET.tostring(locking_configuration, encoding='unicode')
 649
 650         try:
 651             _ = request(data=data)  # type: ignore
 652         except RequestException as e:
 653             raise DashboardException(msg=str(e), component='rgw')