from .enums import (
AuthMode,
CephFSStorageProvider,
+ HostAccess,
JoinSourceType,
LoginAccess,
LoginCategory,
cfg['options'][f'{ceph_vfs}:proxy'] = proxy_val
# 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']['admin users'] = ' '.join(admin_users)
+def _generate_share_hosts_access(
+ share: resources.Share, cfg: Simplified
+) -> None:
+ if not share.hosts_access:
+ return
+ default_access = HostAccess.ALLOW
+ hosts_allow: List[str] = []
+ hosts_deny: List[str] = []
+ for entry in share.hosts_access:
+ if entry.access is HostAccess.DENY:
+ hosts_deny.append(entry.normalized_value)
+ elif entry.access is HostAccess.ALLOW:
+ hosts_allow.append(entry.normalized_value)
+ default_access = HostAccess.DENY
+ else:
+ raise ValueError(f'invalid access type: {entry.access!r}')
+ if default_access is HostAccess.DENY:
+ hosts_deny.append('ALL')
+ if hosts_allow:
+ cfg['options']['hosts allow'] = ', '.join(hosts_allow)
+ if hosts_deny:
+ cfg['options']['hosts deny'] = ', '.join(hosts_deny)
+
+
def _generate_config(
cluster: resources.Cluster,
shares: Iterable[resources.Share],