用实际拒绝结果验证 systemd 服务边界

from pathlib import Path
from tempfile import mkdtemp
import grp,json,os,pwd,socket,subprocess,sys,uuid

def cleanup_steps(steps, primary):
    failures=[]
    for label, action in steps:
        try:action()
        except BaseException as error:
            failures.append(label+': '+type(error).__name__+': '+str(error))
    if failures:
        message='cleanup incomplete or uncertain: '+'; '.join(failures)
        try:print(message,file=sys.stderr)
        except (OSError,ValueError):pass
        if primary is None:raise RuntimeError(message)
    return failures

def remove_file(path):
    try:path.unlink()
    except FileNotFoundError:pass

def require_empty(value):
    if value:raise RuntimeError('residual resources: '+str(value))

suffix=uuid.uuid4().hex[:12];name='xbcatm-hardening-'+suffix
unit=name+'.service';unit_path=Path('/run/systemd/system')/unit
state=Path('/var/lib')/name;marker=Path('/tmp')/(name+'-marker');scratch=Path('/tmp')/(name+'-scratch')
assert os.geteuid()==0,'requires an explicitly prepared disposable systemd test system'
for path in [unit_path,state,marker,scratch]:assert not os.path.lexists(path),path
def account_exists():
    try:pwd.getpwnam(name);return True
    except KeyError:return False
def group_exists():
    try:grp.getgrnam(name);return True
    except KeyError:return False
assert not account_exists() and not group_exists()
def command(args):
    result=subprocess.run(args,capture_output=True,text=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
def ctl(*args):return command(['systemctl',*args])
def delete_account():
    if account_exists():command(['userdel',name])
def delete_group():
    if group_exists():command(['groupdel',name])
root=None;listener=connection=None;owned_files=[];account_attempted=False;unit_attempted=False
# Reserve each ordinary file exclusively and record it before writing or chmod.
def create_file(path,text,mode):
    with path.open('x') as output:
        owned_files.append(path)
        output.write(text)
    path.chmod(mode)
print('fixture:',name,'unit:',unit_path,'state:',state,'marker:',marker,'scratch:',scratch,flush=True)
try:
    root=Path(mkdtemp(prefix=name+'-',dir='/run'));root.chmod(0o755)
    outside=root/'outside';create_file(outside,'initial\n',0o666)
    create_file(marker,'host namespace marker\n',0o644)
    create_file(scratch,'',0o666)
    listener=socket.socket();listener.settimeout(2)
    listener.bind(('127.0.0.1',0));listener.listen(2)
    worker=root/'worker.py'
    create_file(worker,r"""from pathlib import Path
import json,socket,sys
root=Path(__file__).parent;state=Path(sys.argv[1]);marker=Path(sys.argv[2]);scratch=Path(sys.argv[3]);port=int(sys.argv[4]);phase=sys.argv[5]
result={}
try:
    with (root/'outside').open('a') as output:output.write(phase+'\n')
    result['outside_write']=True
except OSError as error:result['outside_write']=False;result['outside_errno']=error.errno
result['host_tmp_visible']=marker.exists()
scratch.write_text(phase);result['tmp_write']=scratch.read_text()==phase
try:
    with socket.create_connection(('127.0.0.1',port),timeout=0.5):pass
    result['network_connect']=True
except OSError as error:result['network_connect']=False;result['network_errno']=error.errno
status=dict(line.split(':',1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line)
result['no_new_privs']=status['NoNewPrivs'].strip()
(state/'result.json').write_text(json.dumps(result))
""",0o644)
    definition=f'''[Service]
Type=oneshot
User={name}
Group={name}
StateDirectory={name}
StateDirectoryMode=0700
TimeoutStartSec=5s
TimeoutStopSec=5s
KillMode=control-group
ExecStart=/usr/bin/python3 {worker} {state} {marker} {scratch} {listener.getsockname()[1]} PHASE
'''
    account_attempted=True
    command(['useradd','--system','--no-create-home','--user-group','--shell','/usr/sbin/nologin',name])
    create_file(unit_path,definition.replace('PHASE','baseline'),0o644)
    unit_attempted=True
    ctl('daemon-reload');ctl('start',unit)
    baseline=json.loads((state/'result.json').read_text())
    assert baseline=={'outside_write':True,'host_tmp_visible':True,'tmp_write':True,'network_connect':True,'no_new_privs':'0'},baseline
    connection,_=listener.accept()
    connection.close();connection=None
    original=outside.read_text()
    settings='ProtectSystem=strict\nPrivateTmp=yes\nPrivateNetwork=yes\nNoNewPrivileges=yes\n'
    unit_path.write_text(definition.replace('PHASE','hardened')+settings)
    command(['systemd-analyze','verify',str(unit_path)])
    ctl('daemon-reload');ctl('start',unit)
    hardened=json.loads((state/'result.json').read_text())
    assert hardened['outside_write'] is False and hardened['outside_errno']==30,hardened
    assert hardened['host_tmp_visible'] is False and hardened['tmp_write'] is True
    assert hardened['network_connect'] is False and hardened['network_errno']==111,hardened
    assert hardened['no_new_privs']=='1'
    assert outside.read_text()==original and scratch.read_text()=='baseline'
    print('baseline: outside write, host tmp visibility, loopback connection permitted; NoNewPrivs=0')
    print('hardened: outside write=EROFS; host tmp hidden; host listener refused; NoNewPrivs=1')
    print('managed state directory and private temporary writes still succeed')
finally:
    primary=sys.exc_info()[1];failures=[];quiescent=not unit_attempted
    # A failed stop does not skip the independently attempted state query or socket close.
    def attempt(label,action):
        try:return action()
        except BaseException as error:
            failures.append(label+': '+type(error).__name__+': '+str(error))
    if unit_attempted:
        attempt('stop '+unit,lambda:ctl('stop',unit))
        status=attempt('unit state',lambda:ctl('show',unit,'-p','ActiveState','-p','MainPID','-p','ControlPID'))
        if status is not None:
            properties=dict(line.split('=',1) for line in status.splitlines() if '=' in line)
            quiescent=properties.get('ActiveState') in ('inactive','failed') and properties.get('MainPID')=='0' and properties.get('ControlPID')=='0'
    if connection is not None:attempt('accepted socket close',connection.close)
    if listener is not None:attempt('listener close',listener.close)
    if quiescent:
        if unit_attempted:
            attempt('reset failed state',lambda:ctl('reset-failed',unit))
        for path in reversed(owned_files):attempt('file '+str(path),lambda path=path:remove_file(path))
        if unit_attempted:attempt('daemon reload',lambda:ctl('daemon-reload'))
        if unit_attempted and os.path.lexists(state):
            if state.is_symlink():failures.append('unexpected state symlink; retained '+str(state))
            else:
                attempt('state result',lambda:remove_file(state/'result.json'))
                attempt('state directory',state.rmdir)
        if account_attempted:
            attempt('account '+name,delete_account)
            attempt('group '+name,delete_group)
        if root is not None:attempt('temporary directory '+str(root),root.rmdir)
        for path in owned_files+([state] if unit_attempted else [])+([root] if root is not None else []):
            attempt('absence '+str(path),lambda path=path:require_empty(str(path) if os.path.lexists(path) else ''))
        if account_attempted:
            attempt('account absence',lambda:require_empty(name if account_exists() else ''))
            attempt('group absence',lambda:require_empty(name if group_exists() else ''))
    else:
        failures.append('unit may still run; retain its unit, account/group, state and files for manual recovery: '+name)
    if failures:
        message='cleanup incomplete or uncertain: '+'; '.join(failures)
        try:print(message,file=sys.stderr)
        except (OSError,ValueError):pass
        if primary is None:raise RuntimeError(message)