import enum
import json
+import shlex
import subprocess
elif load is LoadJSON.ERROR:
return JSONResult(proc.returncode, None, proc.stderr.decode())
return proc
+
+
+def cephadm_enter_cmd(smb_cfg, cluster_id, args, **kwargs):
+ """Run a command inside the primary smbd container for the given
+ cluster_id on the cluster's admin node (derived via smb_cfg).
+ All kwargs are treated as arguments to subprocess.run.
+ """
+ remote_cmd = [
+ 'sudo',
+ f'/home/{smb_cfg.ssh_user}/cephtest/cephadm',
+ 'enter',
+ '-i',
+ f'smb.{cluster_id}',
+ ] + list(args)
+ cmd = [
+ 'ssh',
+ '-oBatchMode=yes',
+ '-oUserKnownHostsFile=/dev/null',
+ '-oStrictHostKeyChecking=no',
+ '-q',
+ f'{smb_cfg.ssh_user}@{smb_cfg.ssh_admin_host}',
+ shlex.join(remote_cmd),
+ ]
+ return subprocess.run(cmd, **kwargs)
--- /dev/null
+import pytest
+
+import cephutil
+import smbutil
+
+
+@pytest.mark.domain
+def test_sid_resolution(smb_cfg):
+ """Verify that rpcclient lookupsids resolves domain user SIDs correctly
+ inside the smbd container, preventing regressions on /run bind mount
+ permissions that break smbd to winbindd communication (tracker#77120).
+ """
+ cluster_id = smbutil.get_shares(smb_cfg)[0]['cluster_id']
+ username = smb_cfg.username
+ password = smb_cfg.password
+
+ result = cephutil.cephadm_enter_cmd(
+ smb_cfg,
+ cluster_id,
+ ['wbinfo', '-n', username],
+ capture_output=True,
+ check=True,
+ )
+ user_sid = result.stdout.decode().split()[0]
+ assert user_sid.startswith('S-'), f'unexpected SID format: {user_sid}'
+
+ auth = f'{username}%{password}'
+ result = cephutil.cephadm_enter_cmd(
+ smb_cfg,
+ cluster_id,
+ [
+ 'rpcclient',
+ 'localhost',
+ '-U',
+ auth,
+ '-c',
+ f'lookupsids {user_sid}',
+ ],
+ capture_output=True,
+ check=True,
+ )
+ output = result.stdout.decode()
+ short_name = username.split('\\')[-1]
+ assert short_name in output, (
+ f'SID resolution failed: {short_name!r} not found in: {output}'
+ )