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 <ssodani@redhat.com>

diff --git a/src/pybind/mgr/smb/external.py b/src/pybind/mgr/smb/external.py
index 412efbb4c0ccbff84a039cc5235bc924413ae9f2..8937411e28d49fafbcbb36c2751290812b37e848 100644 (file)
--- 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 10364024181050973bba31caae90f8034b261e4e..dc72afa880df76b36f3dbf1e4cbd9408ab9e5a25 100644 (file)
--- 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 eda1c02b01ae350be51b422ca4367be3ca5aa41d..aceb776cbfe1284542fec4356b430ab8278eac15 100644 (file)
--- 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(