From: Shweta Sodani Date: Mon, 20 Apr 2026 09:17:35 +0000 (+0530) Subject: mgr/smb: Update handler to process RGW shares, added validation and error handling X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=d317e318592922f803de41b16929f6c7483ebcce;p=ceph.git mgr/smb: Update handler to process RGW shares, added validation and error handling - Add _generate_rgw_share() function for RGW share configuration - Skip CephFS authorization for RGW shares in cluster assembly - Configure Samba VFS with RGW-specific parameters: - rgw_bucket, rgw_user_id, rgw_access_key, rgw_secret_key - Validate bucket accessibility using radosgw-admin - Ensure proper error handling in share creation flow This enables the handler to generate proper Samba configuration for RGW-backed shares along with validation and error handling. Signed-off-by: Shweta Sodani --- diff --git a/src/pybind/mgr/smb/external.py b/src/pybind/mgr/smb/external.py index 412efbb4c0c..8937411e28d 100644 --- a/src/pybind/mgr/smb/external.py +++ b/src/pybind/mgr/smb/external.py @@ -49,6 +49,11 @@ def spec_backup_key(cluster_id: str) -> EntryKey: return (cluster_id, 'spec.smb') +def rgw_credentials_key(cluster_id: str, share_id: str) -> EntryKey: + """Return key identifying RGW credentials for a share in an external store.""" + return (cluster_id, f'rgw-creds.{share_id}.json') + + def tls_credential_key( cluster_id: str, tls_credential_id: str, cred_type: str ) -> EntryKey: diff --git a/src/pybind/mgr/smb/handler.py b/src/pybind/mgr/smb/handler.py index 10364024181..dc72afa880d 100644 --- a/src/pybind/mgr/smb/handler.py +++ b/src/pybind/mgr/smb/handler.py @@ -27,7 +27,7 @@ from ceph.deployment.service_spec import ( ) from ceph.fs.earmarking import EarmarkTopScope -from . import config_store, external, resources +from . import config_store, external, resources, rgw from .enums import ( AuthMode, CephFSStorageProvider, @@ -318,6 +318,7 @@ class ClusterConfigHandler: authorizer: Optional[AccessAuthorizer] = None, orch: Optional[OrchSubmitter] = None, earmark_resolver: Optional[EarmarkResolver] = None, + tool_execer: 'rgw.ToolExecer', ) -> None: self.internal_store = internal_store self.public_store = public_store @@ -332,6 +333,7 @@ class ClusterConfigHandler: if earmark_resolver is None: earmark_resolver = cast(EarmarkResolver, _FakeEarmarkResolver()) self._earmark_resolver = earmark_resolver + self._tool_execer = tool_execer log.info( 'Initialized new ClusterConfigHandler with' f' internal store {self.internal_store!r},' @@ -351,7 +353,7 @@ class ClusterConfigHandler: """ log.debug('applying changes to internal data store') results = ResultGroup() - staging = Staging(self.internal_store) + staging = Staging(self.internal_store, self._tool_execer) try: incoming = order_resources(inputs) for resource in incoming: @@ -639,18 +641,43 @@ class ClusterConfigHandler: assert isinstance(cluster, resources.Cluster) # vols: hold the cephfs volumes our shares touch. some operations are # disabled/skipped unless we touch volumes. - vols = {share.checked_cephfs.volume for share in change_group.shares} + vols = { + share.checked_cephfs.volume + for share in change_group.shares + if share.cephfs is not None + } + # rgw_buckets: hold the RGW buckets our shares touch + rgw_buckets = { + share.rgw.bucket + for share in change_group.shares + if share.rgw is not None + } # save the various object types previous_info = _swap_pending_cluster_info( self.public_store, change_group, - orch_needed=bool(vols and self._orch), + orch_needed=bool((vols or rgw_buckets) and self._orch), ) _save_pending_join_auths(self.priv_store, change_group) _save_pending_users_and_groups(self.priv_store, change_group) _save_pending_tls_credentials(self.priv_store, change_group) + _save_pending_rgw_credentials(self.priv_store, change_group) + rgw_credential_entries = { + share.share_id: change_group.cache[ + external.rgw_credentials_key( + change_group.cluster.cluster_id, share.share_id + ) + ] + for share in change_group.shares + if share.rgw is not None + and share.rgw.access_key_id + and share.rgw.secret_access_key + } cluster_conf = _ClusterConf.assemble( - change_group, self._path_resolver, self._authorizer + change_group, + self._path_resolver, + self._authorizer, + rgw_credential_entries, ) _save_pending_config(self.public_store, cluster_conf) # remove any stray objects @@ -689,6 +716,17 @@ class ClusterConfigHandler: ] for tc in change_group.tls_credentials } + rgw_credential_entries = { + share.share_id: change_group.cache[ + external.rgw_credentials_key( + cluster.cluster_id, share.share_id + ) + ] + for share in change_group.shares + if share.rgw is not None + and share.rgw.access_key_id + and share.rgw.secret_access_key + } ext_ceph_cluster = None if change_group.ext_ceph_clusters: assert len(change_group.ext_ceph_clusters) == 1 @@ -713,9 +751,31 @@ class ClusterConfigHandler: # via orch. This differs from NFS because ganesha embeds the cephx # keys directly in each export definition block while samba needs the # ceph keyring to load keys. + # For RGW shares, we don't need CephFS volumes but still need orchestration. previous_orch = previous_info.get('orch_needed', False) - if self._orch and (vols or previous_orch): + if self._orch and (vols or rgw_buckets or previous_orch): + log.debug( + 'Submitting SMB service spec for cluster %s: ' + 'shares=%d (cephfs_vols=%d, rgw_buckets=%d), ' + 'previous_orch=%s', + cluster.cluster_id, + len(change_group.shares), + len(vols), + len(rgw_buckets), + previous_orch, + ) self._orch.submit_smb_spec(smb_spec) + else: + log.debug( + 'Skipping SMB service orchestration for cluster %s: ' + 'orch_enabled=%s, shares=%d, cephfs_vols=%d, rgw_buckets=%d, previous_orch=%s', + cluster.cluster_id, + bool(self._orch), + len(change_group.shares), + len(vols), + len(rgw_buckets), + previous_orch, + ) def _remove_cluster(self, cluster_id: str) -> None: log.info('Removing cluster: %s', cluster_id) @@ -770,6 +830,8 @@ class _ShareConf: resolver: PathResolver cephx_entity: str ceph_cluster: str + rgw_access_key_uri: Optional[str] = None + rgw_secret_key_uri: Optional[str] = None @dataclasses.dataclass(frozen=True) @@ -785,6 +847,7 @@ class _ClusterConf: change_group: ClusterChangeGroup, default_resolver: PathResolver, authorizer: AccessAuthorizer, + rgw_credential_entries: Dict[str, ConfigEntry], ) -> Self: extcc = None assert isinstance(change_group.cluster, resources.Cluster) @@ -802,25 +865,36 @@ class _ClusterConf: ceph_cluster = 'exo' cephx_entity = checked(extcc.cluster).cephfs_user.name elif change_group.shares: - log.debug('local ceph cluster with shares') - cephx_entity = _cephx_data_entity(change_group.cluster) - # ensure an entity exists with access to the volumes - for share in change_group.shares: - authorizer.authorize_entity( - share.checked_cephfs.volume, cephx_entity + # Check if we have any CephFS shares that need CephX auth + has_cephfs_shares = any( + s.cephfs is not None for s in change_group.shares + ) + if has_cephfs_shares: + log.debug('local ceph cluster with CephFS shares') + cephx_entity = _cephx_data_entity(change_group.cluster) + # ensure an entity exists with access to the volumes (CephFS only) + for share in change_group.shares: + if share.cephfs: + authorizer.authorize_entity( + share.checked_cephfs.volume, cephx_entity + ) + cephadm_data_entity = cephx_entity + else: + log.debug( + 'local ceph cluster with RGW shares only (no CephX auth needed)' ) - cephadm_data_entity = cephx_entity else: log.debug('local cluster without shares: skipping ceph auth') return cls( change_group.cluster, [ - _ShareConf( + _make_share_conf( s, change_group.cluster, resolver, cephx_entity, ceph_cluster, + rgw_credential_entries, ) for s in change_group.shares ], @@ -829,6 +903,80 @@ class _ClusterConf: ) +def _make_share_conf( + s: resources.Share, + cluster: resources.Cluster, + resolver: PathResolver, + cephx_entity: str, + ceph_cluster: str, + rgw_credential_entries: Dict[str, ConfigEntry], +) -> _ShareConf: + creds_entry = rgw_credential_entries.get(s.share_id) + # Get the URI from the ConfigEntry object, not from the data + access_key_uri = ( + f'URI:{creds_entry.uri}:access_key_id' if creds_entry else None + ) + secret_key_uri = ( + f'URI:{creds_entry.uri}:secret_access_key' if creds_entry else None + ) + return _ShareConf( + s, + cluster, + resolver, + cephx_entity, + ceph_cluster, + rgw_access_key_uri=access_key_uri, + rgw_secret_key_uri=secret_key_uri, + ) + + +def _generate_rgw_share( + conf: _ShareConf, +) -> Dict[str, Dict[str, str]]: + """Generate Samba configuration for an RGW-backed share.""" + share = conf.resource + rgw = share.rgw + assert rgw is not None, "RGW storage configuration missing" + + # Use URI-based credential references (already formatted in _make_share_conf) + access_key_uri = conf.rgw_access_key_uri or '' + secret_key_uri = conf.rgw_secret_key_uri or '' + + cfg = { + # smb.conf options + 'options': { + 'path': '/', + 'vfs objects': 'ceph_rgw', + 'ceph_rgw:config_file': '/etc/ceph/ceph.conf', + 'ceph_rgw:keyring_file': '/etc/ceph/ceph.client.admin.keyring', + 'ceph_rgw:bucket': rgw.bucket, + 'ceph_rgw:user_id': rgw.user_id or '', + 'ceph_rgw:access_key': access_key_uri, + 'ceph_rgw:secret_access_key': secret_key_uri, + 'ceph_rgw:debug': 'off', + 'read only': ynbool(share.readonly), + 'browseable': ynbool(share.browseable), + 'kernel share modes': 'no', + 'smbd profiling share': 'yes', + } + } + + if share.comment is not None: + cfg['options']['comment'] = share.comment + if share.max_connections is not None: + cfg['options']['max connections'] = str(share.max_connections) + + # extend share with user+group login access lists + _generate_share_login_control(share, cfg) + _generate_share_hosts_access(share, cfg) + # extend share with custom options + custom_opts = share.cleaned_custom_smb_share_options + if custom_opts: + cfg['options'].update(custom_opts) + cfg['options']['x:ceph:has_custom_options'] = 'yes' + return cfg + + def _generate_share(conf: _ShareConf) -> Dict[str, Dict[str, str]]: share = conf.resource cephx_entity = conf.cephx_entity @@ -1009,7 +1157,12 @@ def _generate_config(conf: _ClusterConf) -> Dict[str, Any]: _set_debug_level(cluster_global_opts, conf) share_configs = { - share.resource.name: _generate_share(share) for share in conf.shares + share.resource.name: ( + _generate_rgw_share(share) + if share.resource.rgw + else _generate_share(share) + ) + for share in conf.shares } instance_features = [] @@ -1309,6 +1462,31 @@ def _save_pending_tls_credentials( change_group.cache_updated_entry(tc_entry) +def _save_pending_rgw_credentials( + store: ConfigStore, + change_group: ClusterChangeGroup, +) -> None: + """Save RGW credentials for shares in the priv store.""" + cluster = change_group.cluster + assert isinstance(cluster, resources.Cluster) + + # Save credentials for each RGW share + for share in change_group.shares: + if share.rgw is not None: + # Only save if credentials are present + if share.rgw.access_key_id and share.rgw.secret_access_key: + ext_key = external.rgw_credentials_key( + cluster.cluster_id, share.share_id + ) + creds_entry = store[ext_key] + creds_data = { + 'access_key_id': share.rgw.access_key_id, + 'secret_access_key': share.rgw.secret_access_key, + } + creds_entry.set(creds_data) + change_group.cache_updated_entry(creds_entry) + + def _save_pending_config( store: ConfigStore, cluster_conf: _ClusterConf, @@ -1350,11 +1528,12 @@ def _store_transaction(store: ConfigStore) -> Iterator[None]: def _has_proxied_vfs(change_group: ClusterChangeGroup) -> bool: - """Return true if any shares in the change group use the new vfs module - with the proxied cephfs library. + """Return true if any CephFS-backed shares in the change group use the + new vfs module with the proxied cephfs library. """ return any( - s.checked_cephfs.provider.expand() + s.cephfs is not None + and s.checked_cephfs.provider.expand() == CephFSStorageProvider.SAMBA_VFS_PROXIED for s in change_group.shares ) diff --git a/src/pybind/mgr/smb/staging.py b/src/pybind/mgr/smb/staging.py index eda1c02b01a..aceb776cbfe 100644 --- a/src/pybind/mgr/smb/staging.py +++ b/src/pybind/mgr/smb/staging.py @@ -15,7 +15,7 @@ import operator from ceph.fs.earmarking import EarmarkTopScope -from . import config_store, resources +from . import config_store, resources, rgw from .enums import ( AuthMode, ConfigNS, @@ -54,8 +54,9 @@ class Staging: the destination store. """ - def __init__(self, store: ConfigStore) -> None: + def __init__(self, store: ConfigStore, tool_exec: rgw.ToolExecer) -> None: self.destination_store = store + self._tool_execer = tool_exec self.incoming: Dict[EntryKey, SMBResource] = {} self.deleted: Dict[EntryKey, SMBResource] = {} self._store_keycache: Set[EntryKey] = set() @@ -358,6 +359,63 @@ def _check_share_resource( msg="no matching cluster id", status={"cluster_id": share.cluster_id}, ) + + # Handle RGW shares - validate bucket existence and auto-fetch credentials + if share.rgw is not None: + # Validate bucket exists + if not rgw.validate_rgw_bucket( + staging._tool_execer, share.rgw.bucket + ): + raise ErrorResult( + share, + msg=f"RGW bucket '{share.rgw.bucket}' does not exist or is not accessible", + ) + + # Auto-fetch credentials if not provided + if ( + not share.rgw.user_id + or not share.rgw.access_key_id + or not share.rgw.secret_access_key + ): + try: + log.debug( + f"Auto-fetching RGW credentials for bucket {share.rgw.bucket}" + ) + ( + fetched_user_id, + access_key, + secret_key, + ) = rgw.fetch_rgw_credentials( + staging._tool_execer, + share.rgw.bucket, + share.rgw.user_id or '', + ) + # Update the share's RGW storage with fetched credentials + share.rgw = resources.RGWStorage( + bucket=share.rgw.bucket, + user_id=fetched_user_id, + access_key_id=access_key, + secret_access_key=secret_key, + ) + log.debug( + f"Successfully fetched credentials for user {fetched_user_id}" + ) + except ValueError as e: + raise ErrorResult( + share, + msg=f"Failed to fetch RGW credentials: {str(e)}", + ) + + name_used_by = _share_name_in_use(staging, share) + if name_used_by: + raise ErrorResult( + share, + msg="share name already in use", + status={"conflicting_share_id": name_used_by}, + ) + return + + # Handle CephFS shares assert share.cephfs is not None try: volpath = path_resolver.resolve_exists(