314 lines
15 KiB
Python
314 lines
15 KiB
Python
#!/usr/bin/python3
|
|
"""Actual native Tor worker + production watchdog/systemctl in disposable PID1.
|
|
|
|
Exercises the finite role independently. Never invokes or replaces admit().
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
BASE = Path('/opt/c01-fixture')
|
|
OUT = Path('/var/lib/tor-qualification')
|
|
sys.path.insert(0, str(BASE / 'scripts'))
|
|
import c01_supervisor as supervisor
|
|
import c01_tor
|
|
|
|
|
|
def run(*args, timeout=120):
|
|
p = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
|
|
assert p.returncode == 0, (args, p.returncode, p.stdout, p.stderr)
|
|
return p.stdout
|
|
|
|
|
|
def record(name, value):
|
|
(OUT / name).write_text(json.dumps(value, indent=2) + '\n')
|
|
|
|
|
|
def snapshot():
|
|
paths = list(Path('/var/lib/tor').glob('hidden_service_*'))
|
|
paths += [item for directory in paths for item in directory.iterdir()]
|
|
paths += [Path('/var/lib/archipelago/runtime/lnd/wallet-canary')]
|
|
result = {}
|
|
for path in paths:
|
|
m = path.lstat()
|
|
result[str(path)] = dict(inode=m.st_ino, device=m.st_dev, uid=m.st_uid,
|
|
gid=m.st_gid, mode=m.st_mode,
|
|
sha256=hashlib.sha256(path.read_bytes()).hexdigest() if path.is_file() else None)
|
|
return result
|
|
|
|
|
|
class Session:
|
|
def __init__(self):
|
|
self.lock = supervisor.LaunchLock()
|
|
self.owner = supervisor.Supervisor(self.lock)
|
|
self.listener = c01_tor.listener()
|
|
address = self.listener.getsockname()
|
|
parent = os.getpid()
|
|
def app(control):
|
|
supervisor.lifecycle.drop_credentials(65001, 65001, parent)
|
|
channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
channel.connect(address)
|
|
supervisor.packet(control, b'CONNECTED')
|
|
while True:
|
|
request = control.recv(4096)
|
|
if not request:
|
|
return
|
|
if request == b'FIXTURE_TRANSFER_SOCKET':
|
|
import array
|
|
control.sendmsg([b'SOCKET'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS,
|
|
array.array('i', [channel.fileno()]))])
|
|
continue
|
|
channel.send(request)
|
|
reply = channel.recv(4096)
|
|
if not reply:
|
|
return
|
|
supervisor.packet(control, reply)
|
|
self.app = self.owner.children.spawn('application', app)
|
|
self.owner.wait(self.app, b'CONNECTED', 65001, 65001, supervisor.now()+2_000_000_000)
|
|
accepted, _ = self.listener.accept()
|
|
runtime = os.open('/var/lib/archipelago/runtime', os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
|
|
self.dirs = c01_tor.capabilities(runtime)
|
|
os.close(runtime)
|
|
self.executable = os.open('/usr/local/bin/archipelago', os.O_RDONLY | os.O_CLOEXEC)
|
|
self.worker = c01_tor.spawn(self.owner, self.executable, accepted,
|
|
dict(pid=self.app.pid, uid=65001, gid=65001), self.dirs,
|
|
supervisor.native, supervisor.packet, {'PATH':'/usr/bin:/usr/sbin'})
|
|
end = supervisor.now()+2_000_000_000
|
|
while self.owner.tor_watch.state == 'starting' and supervisor.now() < end and self.owner.tick():
|
|
time.sleep(.01)
|
|
assert self.owner.tor_watch.state == 'idle', self.owner.failure
|
|
self.pid = self.worker.pid
|
|
self.group = str(self.worker.group.path)
|
|
|
|
def request(self, operation, sequence=None):
|
|
request = 'Status' if sequence is None else {'Apply':dict(sequence=sequence, operation=operation)}
|
|
supervisor.packet(self.app.channel, request)
|
|
return json.loads(self.owner.receive(self.app, 65001, 65001, supervisor.now()+65_000_000_000))
|
|
|
|
def close(self):
|
|
self.owner.stop('fixture retirement')
|
|
end = time.monotonic()+10
|
|
while self.owner.children.children and time.monotonic()<end:
|
|
self.owner.tick()
|
|
time.sleep(.01)
|
|
assert not self.owner.children.children, 'unretired Tor/application roles'
|
|
results = dict(self.owner.children.results)
|
|
self.owner.children.close()
|
|
os.close(self.owner.wake_read)
|
|
os.close(self.owner.wake_write)
|
|
import signal
|
|
signal.set_wakeup_fd(self.owner.prior_wakeup)
|
|
self.lock.close()
|
|
self.listener.close()
|
|
for fd in [self.executable, *self.dirs]:
|
|
os.close(fd)
|
|
assert not Path(self.group).exists()
|
|
return results
|
|
|
|
def root_sender_refused(self):
|
|
# Deliberately transfer the app endpoint to root. Possessing the socket
|
|
# must not replace exact per-packet nonroot PID/UID authentication.
|
|
import array
|
|
supervisor.packet(self.app.channel, b'FIXTURE_TRANSFER_SOCKET')
|
|
self.app.channel.settimeout(2)
|
|
raw, anc, flags, _ = self.app.channel.recvmsg(64, socket.CMSG_SPACE(12)+socket.CMSG_SPACE(4), socket.MSG_CMSG_CLOEXEC)
|
|
assert raw == b'SOCKET' and not flags & ~socket.MSG_CMSG_CLOEXEC
|
|
rights = array.array('i')
|
|
for level, kind, data in anc:
|
|
if (level,kind) == (socket.SOL_SOCKET,socket.SCM_RIGHTS):
|
|
rights.frombytes(data)
|
|
assert len(rights) == 1
|
|
forged = socket.socket(fileno=rights[0])
|
|
try:
|
|
forged.send(json.dumps({'Apply':dict(sequence=3,operation='Reload')}).encode())
|
|
end = supervisor.now()+2_000_000_000
|
|
while supervisor.now()<end and self.owner.tick():
|
|
time.sleep(.01)
|
|
assert self.owner.tor_watch.state == 'failed', 'root sender was accepted as the app'
|
|
finally:
|
|
forged.close()
|
|
|
|
|
|
def role_tests():
|
|
OUT.mkdir(mode=0o700, exist_ok=True)
|
|
# Execute the production embedded module table too. The held real launch
|
|
# lock must stop it before acquiring another Children owner or admission.
|
|
guard = supervisor.LaunchLock()
|
|
try:
|
|
embedded = subprocess.run(['/usr/local/bin/archipelago','c01-supervisor-v3'],
|
|
capture_output=True,text=True,timeout=20)
|
|
finally:
|
|
guard.close()
|
|
record('embedded-supervisor.json', dict(exit=embedded.returncode, stderr=embedded.stderr))
|
|
assert embedded.returncode != 0 and 'BlockingIOError' in embedded.stderr and '<embedded:c01_supervisor>' in embedded.stderr, embedded.stderr
|
|
healthy = Path('/etc/tor/torrc').read_text()
|
|
original = snapshot()
|
|
before_pid = run('systemctl','show','tor@default.service','-p','MainPID','--value').strip()
|
|
Path('/etc/tor/torrc').write_text(healthy.replace('SocksPort 9050\n',
|
|
'SocksPort 9050\nSocksPort 192.0.2.88:9050\n'))
|
|
session = Session()
|
|
try:
|
|
record('worker.json', dict(pid=session.pid, group=session.group,
|
|
memory_max=Path(session.group+'/memory.max').read_text(),
|
|
swap_max=Path(session.group+'/memory.swap.max').read_text(),
|
|
tasks_max=Path(session.group+'/pids.max').read_text()))
|
|
assert session.request(None)['sequence'] == 0
|
|
reply = session.request('Heal', 1)
|
|
assert reply['outcome'] == dict(changed=True,reloaded=True,active=True,exported=1), reply
|
|
assert Path('/etc/tor/torrc').read_text() == healthy
|
|
assert run('systemctl','show','tor@default.service','-p','MainPID','--value').strip() == before_pid
|
|
assert Path('/var/lib/archipelago/runtime/tor-hostnames/archipelago').read_bytes() == Path('/var/lib/tor/hidden_service_archipelago/hostname').read_bytes()
|
|
assert snapshot() == original
|
|
record('repair.json', reply)
|
|
# Same durable sequence is a cached completion, not a second reload.
|
|
assert session.request('Heal', 1) == reply
|
|
run('systemctl', 'stop', 'tor@default.service')
|
|
stopped = session.request('Reload', 2)
|
|
assert stopped['outcome']['active'] is False and stopped['outcome']['reloaded'] is False
|
|
assert run('systemctl','show','tor@default.service','-p','ActiveState','--value').strip() == 'inactive'
|
|
assert snapshot() == original
|
|
record('stopped.json', stopped)
|
|
finally:
|
|
record('first-retirement.json', session.close())
|
|
# Fresh direct native child recovers the durable sequence; no new domain.
|
|
session = Session()
|
|
try:
|
|
status = session.request(None)
|
|
assert status['sequence'] == 2 and not status['uncertain']
|
|
record('recovered.json', status)
|
|
finally:
|
|
record('recovery-retirement.json', session.close())
|
|
complete = Path('/var/lib/archipelago-c01/tor/state').read_bytes()
|
|
session = Session()
|
|
try:
|
|
session.root_sender_refused()
|
|
assert Path('/var/lib/archipelago-c01/tor/state').read_bytes() == complete
|
|
assert snapshot() == original
|
|
record('wrong-sender.json', dict(refused=True,journal_preserved=True))
|
|
finally:
|
|
record('wrong-sender-retirement.json', session.close())
|
|
session = Session()
|
|
try:
|
|
try:
|
|
session.request({'Reload':{'path':'/etc/shadow','argv':['start']}},3)
|
|
except (ValueError,OSError,json.JSONDecodeError):
|
|
pass
|
|
else:
|
|
raise AssertionError('nonfinite Tor request accepted')
|
|
assert Path('/var/lib/archipelago-c01/tor/state').read_bytes() == complete
|
|
assert snapshot() == original
|
|
record('nonfinite-request.json',dict(refused=True,journal_preserved=True))
|
|
finally:
|
|
record('nonfinite-retirement.json',session.close())
|
|
run('systemctl', 'start', 'tor@default.service')
|
|
# Make the real PID1 reload job exceed the production sixty-second budget.
|
|
dropin = Path('/etc/systemd/system/tor@default.service.d')
|
|
dropin.mkdir()
|
|
(dropin/'timeout.conf').write_text('[Service]\nExecReload=\nExecReload=/usr/bin/sleep 90\nTimeoutStartSec=120\n')
|
|
run('systemctl','daemon-reload')
|
|
session = Session()
|
|
started = supervisor.now()
|
|
try:
|
|
try:
|
|
session.request('Reload', 3)
|
|
except (ValueError, OSError, json.JSONDecodeError):
|
|
pass
|
|
else:
|
|
raise AssertionError('timed-out mutation reported success')
|
|
elapsed = supervisor.now()-started
|
|
assert 59_000_000_000 <= elapsed <= 65_000_000_000, elapsed
|
|
assert session.owner.tor_watch.state == 'failed'
|
|
intent = json.loads(Path('/var/lib/archipelago-c01/tor/state').read_text())
|
|
assert intent['sequence'] == 3 and intent['outcome'] is None
|
|
assert snapshot() == original
|
|
record('timeout.json', dict(elapsed_ns=elapsed, reason=session.owner.tor_watch.reason,
|
|
journal=intent, service_state=run('systemctl','show','tor@default.service','-p','ActiveState','-p','SubState')))
|
|
finally:
|
|
record('timeout-retirement.json', session.close())
|
|
# No replay after fresh owner startup, even though PID1 may still own the
|
|
# original service job. Reconciliation is external; this fixture stops it.
|
|
session = Session()
|
|
try:
|
|
try:
|
|
status = session.request(None)
|
|
assert status['uncertain'] and status['sequence'] == 3
|
|
except (ValueError, OSError, json.JSONDecodeError):
|
|
assert session.owner.tor_watch.state == 'failed'
|
|
assert json.loads(Path('/var/lib/archipelago-c01/tor/state').read_text()) == intent
|
|
finally:
|
|
record('uncertain-recovery-retirement.json', session.close())
|
|
run('systemctl','stop','tor@default.service')
|
|
assert snapshot() == original
|
|
record('preservation.json', dict(before=original, after=snapshot()))
|
|
record('result.json', dict(exit=0, native_role=True, production_systemctl=True,
|
|
admission_bypassed=False, app_ready=False))
|
|
|
|
|
|
def setup():
|
|
assert os.geteuid() == 0 and Path('/run/.containerenv').exists()
|
|
assert Path('/proc/1/comm').read_text().strip() == 'systemd'
|
|
assert os.environ.get('C01_TOR_PID1') == '1'
|
|
OUT.mkdir(mode=0o700)
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location('native_boundaries', BASE/'tests/regression/c01-normal-systemd.py')
|
|
probes = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(probes)
|
|
probes.EVIDENCE = OUT/'abi'
|
|
probes.EVIDENCE.mkdir()
|
|
for selector, label, expected, slots, passcred in [
|
|
('c01-managed-v6','missing-fd12','required Tor lifecycle FD12 missing',(),False),
|
|
('c01-managed-v6','missing-fd11','required network settings FD11 missing',(12,),False),
|
|
('c01-managed-v6','missing-fd10','required D09 FD10 missing',(11,12),False),
|
|
('c01-managed-v6','foreign-fd10','D09 FD10 type/passcred',(10,11,12),False),
|
|
('c01-managed-v6','parent-not-d09','FD10 must name independent D09 creator',(10,11,12),True),
|
|
*[(f'c01-managed-v{v}',f'obsolete-v{v}','obsolete protected application ABI',(),False) for v in (3,4,5)],
|
|
]:
|
|
probes.native_boundary(selector,label,expected,slots,passcred)
|
|
for path in ['/etc/tor','/var/lib/tor','/var/lib/archipelago-c01/tor',
|
|
'/var/lib/archipelago/runtime/tor-hostnames','/var/lib/archipelago/runtime/lnd']:
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
# Purely synthetic preserved data, never imported from an installed node.
|
|
assert not list(Path('/var/lib/tor').iterdir()), 'fixture image must not carry Tor identities'
|
|
Path('/var/lib/archipelago/runtime/lnd/wallet-canary').write_bytes(b'synthetic-wallet-no-funds')
|
|
uninstalled = Path('/var/lib/tor/hidden_service_uninstalled')
|
|
uninstalled.mkdir(mode=0o700)
|
|
(uninstalled/'hs_ed25519_secret_key').write_bytes(b'synthetic-retained-private-key')
|
|
config = ('# Auto-generated by Archipelago — do not edit manually\n'
|
|
'User debian-tor\nDataDirectory /var/lib/tor\nDisableNetwork 1\nSocksPort 9050\n'
|
|
'SocksPolicy accept 127.0.0.1/32\nSocksPolicy accept 192.0.2.0/24\nSocksPolicy reject *\n'
|
|
'Log notice stdout\nHiddenServiceDir /var/lib/tor/hidden_service_archipelago\n'
|
|
'HiddenServicePort 80 127.0.0.1:80\n')
|
|
Path('/etc/tor/torrc').write_text(config)
|
|
Path('/etc/systemd/system/tor@default.service').write_text('[Service]\nType=simple\n'
|
|
'ExecStart=/usr/bin/tor -f /etc/tor/torrc\nExecReload=/bin/kill -HUP $MAINPID\n'
|
|
'Restart=no\nNoNewPrivileges=yes\n')
|
|
run('systemctl','daemon-reload')
|
|
run('systemctl','start','tor@default.service')
|
|
end = time.monotonic()+20
|
|
while not Path('/var/lib/tor/hidden_service_archipelago/hostname').exists() and time.monotonic()<end:
|
|
time.sleep(.05)
|
|
assert Path('/var/lib/tor/hidden_service_archipelago/hostname').exists()
|
|
# Tor signal readiness is separate from key creation.
|
|
time.sleep(1)
|
|
run('ip', 'netns', 'add', 'archipelago-managed')
|
|
run('systemd-run','--unit=tor-native-fixture','--wait','--pipe',
|
|
'-p','Delegate=yes','-p','DelegateSubgroup=supervisor','-p','NoNewPrivileges=yes',
|
|
'-p','MemoryMax=512M','-p','MemorySwapMax=0','-p','CPUQuota=200%',
|
|
'-p','NetworkNamespacePath=/run/netns/archipelago-managed',
|
|
'python3','-B',str(Path(__file__)),'roles', timeout=180)
|
|
(OUT/'tor-journal.log').write_text(run('journalctl','-u','tor@default.service','--no-pager'))
|
|
assert 'Received reload signal' in (OUT/'tor-journal.log').read_text()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if sys.argv[1:] == ['roles']:
|
|
role_tests()
|
|
else:
|
|
setup()
|