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

from pathlib import Path
from tempfile import TemporaryDirectory
import json,shutil,socket,subprocess,uuid

suffix=uuid.uuid4().hex[:8];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 not unit_path.exists() and not state.exists() and not marker.exists()
def ctl(*args):
    result=subprocess.run(['systemctl',*args],capture_output=True,text=True,timeout=10)
    assert result.returncode==0,(args,result.stderr)
    return result.stdout.strip()
with TemporaryDirectory(prefix=name+'-',dir='/run') as directory, socket.socket() as listener:
    root=Path(directory);root.chmod(0o755);outside=root/'outside';outside.write_text('initial\n');outside.chmod(0o666)
    marker.write_text('host namespace marker\n');listener.bind(('127.0.0.1',0));listener.listen(2)
    worker=root/'worker.py'
    worker.write_text(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))
""")
    definition=f'''[Service]
Type=oneshot
User={name}
Group={name}
StateDirectory={name}
StateDirectoryMode=0700
ExecStart=/usr/bin/python3 {worker} {state} {marker} {scratch} {listener.getsockname()[1]} PHASE
'''
    subprocess.run(['useradd','--system','--no-create-home','--user-group','--shell','/usr/sbin/nologin',name],check=True)
    try:
        unit_path.write_text(definition.replace('PHASE','baseline'))
        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()
        original=outside.read_text()
        settings='ProtectSystem=strict\nPrivateTmp=yes\nPrivateNetwork=yes\nNoNewPrivileges=yes\n'
        unit_path.write_text(definition.replace('PHASE','hardened')+settings)
        subprocess.run(['systemd-analyze','verify',str(unit_path)],check=True)
        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:
        ctl('stop',unit)
        if ctl('show',unit,'-p','ActiveState','--value')=='failed':ctl('reset-failed',unit)
        unit_path.unlink(missing_ok=True);ctl('daemon-reload')
        marker.unlink(missing_ok=True);scratch.unlink(missing_ok=True)
        if state.exists():shutil.rmtree(state)
        subprocess.run(['userdel',name],check=True)