74 lines
4.1 KiB
Python
74 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise firewall sandbox and failure dependency using unique transient units.
|
|
|
|
Requires root. All nft commands run in systemd PrivateNetwork; host rules and
|
|
installed units are untouched. Only fresh /run fixtures and transient units exist.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
import firewall
|
|
|
|
HERE=Path(__file__).resolve().parent
|
|
|
|
def run(args, check=True):
|
|
result=subprocess.run(args,capture_output=True,text=True,timeout=60)
|
|
if check and result.returncode: raise RuntimeError(result.stdout+result.stderr)
|
|
return result
|
|
|
|
def main():
|
|
if os.geteuid()!=0: raise SystemExit('root required for transient sandbox fixtures')
|
|
units=[]
|
|
with tempfile.TemporaryDirectory(prefix='archy-firewall-unit-',dir='/run') as temp:
|
|
root=Path(temp); root.chmod(0o755)
|
|
try:
|
|
os.umask(0o022)
|
|
for name,data in firewall.render(json.loads((HERE/'app-ports.json').read_text())).items():
|
|
path=root/name; path.parent.mkdir(parents=True,exist_ok=True); path.write_bytes(data)
|
|
shutil.copyfile(HERE/'firewall.py',root/'firewall.py')
|
|
(root/'firewall.lock').touch(mode=0o600)
|
|
hostns=os.readlink('/proc/self/ns/net')
|
|
guard=root/'guard.py'
|
|
guard.write_text('import os,runpy\nassert os.readlink("/proc/self/ns/net") != '+repr(hostns)+'\nrunpy.run_path('+repr(str(root/'firewall.py'))+',run_name="__main__")\n')
|
|
def launch(unit, fail=False):
|
|
units.append(unit)
|
|
props={'Type':'oneshot','PrivateNetwork':'yes','NoNewPrivileges':'yes',
|
|
'ProtectSystem':'strict','ProtectHome':'yes','PrivateTmp':'yes','PrivateDevices':'yes',
|
|
'ReadWritePaths':'/run','CapabilityBoundingSet':'CAP_NET_ADMIN',
|
|
'RestrictAddressFamilies':'AF_UNIX AF_NETLINK','MemoryMax':'64M','TasksMax':'8',
|
|
'BindPaths':f'{root}/firewall.lock:/run/archipelago-host-firewall.lock',
|
|
'BindReadOnlyPaths':f'{root}/etc/archipelago:/etc/archipelago {root}/etc/fips:/etc/fips'}
|
|
args=['systemd-run','--unit='+unit,'--wait','--pipe']
|
|
for key,value in props.items(): args+=['--property='+key+'='+value]
|
|
args+=['/usr/bin/python3','-I','-B',str(guard)]
|
|
return run(args,check=not fail)
|
|
good='archy-policy-good-'+uuid.uuid4().hex
|
|
launch(good)
|
|
# Malformed baseline must fail startup; a dependent listener surrogate
|
|
# must never start, with exactly the provisioner's Requires/After model.
|
|
(root/'etc/fips/fips.d/70-invalid.nft').write_text('invalid nft syntax\n')
|
|
bad='archy-policy-bad-'+uuid.uuid4().hex
|
|
assert launch(bad,True).returncode!=0
|
|
dependent='archy-policy-dependent-'+uuid.uuid4().hex; units.append(dependent)
|
|
marker=root/'listener-started'
|
|
result=run(['systemd-run','--unit='+dependent,'--wait','--pipe',
|
|
'--property=Requires='+bad+'.service','--property=After='+bad+'.service',
|
|
'/usr/bin/touch',str(marker)],False)
|
|
# systemd-run may return 0 for a job canceled before ExecStart.
|
|
# Require the failed prerequisite and the manager's dependency-failure
|
|
# journal, as well as the absent listener side effect.
|
|
status=run(['systemctl','show',bad+'.service','--property=Result','--value']).stdout.strip()
|
|
journal=run(['journalctl','-u',dependent+'.service','--no-pager','-n','20']).stdout
|
|
print(journal,flush=True)
|
|
assert status=='exit-code' and not marker.exists() and 'dependency' in journal.lower(), 'missing failed dependency evidence'
|
|
print('PASS: actual systemd sandbox loads isolated nft; failed firewall prevents dependent startup')
|
|
finally:
|
|
for unit in reversed(units):
|
|
run(['systemctl','stop',unit],False); run(['systemctl','reset-failed',unit],False)
|
|
|
|
if __name__=='__main__': main()
|