同一账号的会话共享 systemd 用户管理器

from pathlib import Path
from tempfile import TemporaryDirectory
import os,pwd,subprocess,time,uuid

name='xbcatm-user-'+uuid.uuid4().hex[:8];sessions=[name+'-a.service',name+'-b.service'];unit='field-worker.service'
def run(*args):
    result=subprocess.run(args,capture_output=True,text=True,timeout=20)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def wait_for(check,seconds=5):
    deadline=time.monotonic()+seconds
    while not check():
        assert time.monotonic()<deadline,'state transition timed out';time.sleep(0.05)
with TemporaryDirectory(prefix=name+'-',dir='/run') as directory:
    root=Path(directory);root.chmod(0o755);home=root/'home';home.mkdir()
    run('useradd','--no-create-home','--user-group','--home-dir',str(home),'--shell','/bin/sh',name)
    account=pwd.getpwnam(name);uid=account.pw_uid;gid=account.pw_gid;manager=f'user@{uid}.service'
    def userctl(*args):
        return run('setpriv','--reuid='+str(uid),'--regid='+str(gid),'--init-groups',
                   'env',f'XDG_RUNTIME_DIR=/run/user/{uid}',f'DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/{uid}/bus',
                   'systemctl','--user',*args)
    def manager_value(key):return run('systemctl','show',manager,'-p',key,'--value')
    try:
        units=home/'.config/systemd/user';units.mkdir(parents=True)
        (home/'worker.py').write_text("from pathlib import Path\nimport os,time\np=Path(__file__).with_name('worker-pid');t=p.with_suffix('.tmp');t.write_text(str(os.getpid()));t.replace(p)\ntime.sleep(60)\n")
        (units/unit).write_text('[Service]\nType=exec\nExecStart=/usr/bin/python3 %h/worker.py\n[Install]\nWantedBy=default.target\n')
        for path in [home,*home.rglob('*')]:os.chown(path,uid,gid)
        for session in sessions:
            run('systemd-run','--quiet','--unit='+session,'--property=Type=simple',
                '--property=User='+name,'--property=PAMName=login','/usr/bin/sleep','60')
        wait_for(lambda:Path(f'/run/user/{uid}/bus').exists())
        wait_for(lambda:len(run('loginctl','show-user',name,'-p','Sessions','--value').split())==2)
        assert run('loginctl','show-user',name,'-p','Linger','--value')=='no'
        initial=manager_value('MainPID');assert initial!='0'
        userctl('enable','--now',unit);wait_for(lambda:(home/'worker-pid').exists())
        worker_pid=(home/'worker-pid').read_text()
        assert userctl('show',unit,'-p','MainPID','--value')==worker_pid
        run('systemctl','stop',sessions[0])
        wait_for(lambda:len(run('loginctl','show-user',name,'-p','Sessions','--value').split())==1)
        assert manager_value('MainPID')==initial and userctl('show',unit,'-p','MainPID','--value')==worker_pid
        print('two PAM sessions share one user manager; closing one leaves its service unchanged')
        run('systemctl','stop',sessions[1])
        wait_for(lambda:manager_value('ActiveState')=='inactive',seconds=20)
        assert not Path('/proc/'+worker_pid).exists()
        print('after the final session and configured stop delay, non-lingering manager and worker stop')
        print('controllers use setpriv and the actual user bus; they do not create extra PAM sessions')
    finally:
        for session in sessions:
            if run('systemctl','show',session,'-p','LoadState','--value')!='not-found':run('systemctl','stop',session)
        run('systemctl','stop',manager,f'user-runtime-dir@{uid}.service')
        run('userdel',name)