232 lines
11 KiB
Python
232 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Real authority executable and credentials; synthetic UTC/platform fixture.
|
|
|
|
Runs inside the startup fixture's PID/network/filesystem isolation ONLY. This
|
|
does not qualify suspend, authenticate real UTC, install services or admit nodes.
|
|
"""
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import traceback
|
|
from unittest.mock import patch
|
|
|
|
assert os.geteuid() == 0 and os.getppid() == 1
|
|
assert os.environ.get('ARCHY_TEST_ISOLATED') == '1'
|
|
assert os.environ.get('ARCHY_TEST_C01_STARTUP') == '1'
|
|
for directory in ('/opt/archipelago', '/etc/archipelago'):
|
|
fields = [line.split() for line in Path('/proc/self/mountinfo').read_text().splitlines()]
|
|
assert any(row[4] == directory and row[row.index('-') + 1] == 'tmpfs' for row in fields)
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
|
|
import c01_nts as nts
|
|
import c01_recovery as recovery
|
|
|
|
APP, AUTHORITY = 65534, 65533
|
|
BINDING = {'local_identity': [1] * 32, 'incarnation': [2] * 32, 'reader_schema': 2, 'writer_schema': 2}
|
|
binary = Path(sys.argv[1]).resolve(strict=True)
|
|
|
|
|
|
def drop(uid):
|
|
os.setgroups([])
|
|
os.setresgid(uid, uid, uid)
|
|
os.setresuid(uid, uid, uid)
|
|
|
|
|
|
def close_except(keep):
|
|
for text in os.listdir('/proc/self/fd'):
|
|
fd = int(text)
|
|
if fd > 2 and fd not in keep:
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass # Directory enumeration's own descriptor is already closed.
|
|
|
|
|
|
def receive(control):
|
|
control.settimeout(10)
|
|
raw, _, flags, _ = control.recvmsg(8192)
|
|
assert raw and not flags, 'fixture control unavailable/oversized'
|
|
return json.loads(raw)
|
|
|
|
|
|
def credentials(channel):
|
|
return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
|
|
|
|
def clock_observation():
|
|
reading = nts.clock_reading()
|
|
return {'platform': 'Linux', 'wall': recovery.duration(reading.wall),
|
|
'before': recovery.duration(reading.before), 'after': recovery.duration(reading.after),
|
|
'boot_id': list(bytes.fromhex(recovery.boot_id()))}
|
|
|
|
|
|
def transact(channel, sequence, operation):
|
|
request = {'version': 4, 'binding': BINDING, 'session': [3] * 32,
|
|
'sequence': sequence, 'operation': operation}
|
|
raw = json.dumps(request, separators=(',', ':')).encode()
|
|
channel.send(raw)
|
|
receipt = receive(channel)
|
|
assert receipt['request_sha256'] == list(hashlib.sha256(raw).digest())
|
|
assert receipt['binding'] == BINDING and receipt['session'] == [3] * 32
|
|
assert receipt['sequence'] == sequence and receipt['version'] == 4
|
|
return receipt['reply']
|
|
|
|
|
|
with tempfile.TemporaryDirectory(prefix='c01-recovery-') as temporary:
|
|
root = Path(temporary); root.chmod(0o755)
|
|
executable = root / 'archipelago'
|
|
shutil.copyfile(binary, executable); executable.chmod(0o755)
|
|
for case in ('accepted-once', 'wrong-peer', 'stale-observation'):
|
|
location = root / case; location.mkdir(mode=0o755)
|
|
data = location / 'data'; data.mkdir(mode=0o700); os.chown(data, APP, APP)
|
|
store = location / 'authority'; store.mkdir(mode=0o700); os.chown(store, AUTHORITY, AUTHORITY)
|
|
key_path = location / 'authority.key'; key_path.write_bytes(bytes([9] * 32)); key_path.chmod(0o600)
|
|
key_fd = os.open(key_path, os.O_RDONLY | os.O_CLOEXEC)
|
|
key_metadata = os.fstat(key_fd)
|
|
address = '\0c01-recovery-' + os.urandom(16).hex()
|
|
children = set()
|
|
controls = []
|
|
|
|
def spawn(function, keep=()):
|
|
parent, child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
try:
|
|
close_except({child.fileno(), *keep})
|
|
function(child)
|
|
except BaseException:
|
|
traceback.print_exc()
|
|
os._exit(111)
|
|
os._exit(0)
|
|
child.close()
|
|
children.add(pid); controls.append(parent)
|
|
return pid, parent
|
|
|
|
def finish(pid, expected):
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
done, status = os.waitpid(pid, os.WNOHANG)
|
|
if done:
|
|
children.remove(pid)
|
|
code = os.waitstatus_to_exitcode(status)
|
|
assert code == expected, (case, pid, code, expected)
|
|
return
|
|
time.sleep(0.01)
|
|
raise AssertionError('fixture child did not finish')
|
|
|
|
def authority(control):
|
|
drop(AUTHORITY)
|
|
listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
listener.bind(address); listener.listen(2); listener.settimeout(5)
|
|
control.send(b'{"listening":true}')
|
|
connections = [listener.accept()[0], listener.accept()[0]]
|
|
app = next(s for s in connections if credentials(s)[1] == APP)
|
|
admin = next(s for s in connections if credentials(s)[1] == AUTHORITY)
|
|
control.send(json.dumps({'app': credentials(app), 'admin': credentials(admin)}).encode())
|
|
copies = [fcntl.fcntl(fd, fcntl.F_DUPFD_CLOEXEC, 20)
|
|
for fd in (control.fileno(), app.fileno(), admin.fileno(), key_fd)]
|
|
for slot, fd in zip((3, 4, 5, 9), copies):
|
|
os.dup2(fd, slot, inheritable=True)
|
|
os.closerange(6, 9); os.closerange(10, 1048576)
|
|
os.execve(executable, [str(executable), 'c01-authority-v3'],
|
|
{'PATH': '/usr/bin:/bin', 'HOME': str(location), 'RUST_MIN_STACK': '16777216'})
|
|
|
|
def application(control):
|
|
drop(APP)
|
|
channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
channel.settimeout(5); channel.connect(address)
|
|
action = receive(control)
|
|
if action == {'exercise': False}:
|
|
return # Explicit fixture cancellation after refused admission.
|
|
assert action == {'exercise': True}
|
|
assert transact(channel, 1, {'operation': 'Load'}) == {'result': 'Loaded', 'anchor': None}
|
|
first = transact(channel, 2, {'operation': 'Recover', 'minimum': 0, 'observation': clock_observation()})
|
|
assert first['result'] == 'Clock'
|
|
assert first['recovery']['domain'] == [7] * 16
|
|
assert first['recovery']['wall'] == recovery.duration(approval._observation.utc_ns)
|
|
assert first['recovery']['monotonic'] == recovery.duration(approval._observation.boottime_ns)
|
|
assert first['recovery']['uncertainty'] == recovery.duration(approval._observation.observation_error_ns)
|
|
assert first['recovery']['max_elapsed_error_ppb'] == platform.max_elapsed_error_ppb
|
|
# A fresh transport request still cannot renew consumed time authority.
|
|
assert transact(channel, 3, {'operation': 'Recover', 'minimum': 0,
|
|
'observation': clock_observation()}) == {'result': 'Refused'}
|
|
control.send(b'{"accepted_once":true,"replay_refused":true}')
|
|
|
|
try:
|
|
# Explicit SYNTHETIC time fixture. Only collect() is substituted;
|
|
# binding, platform identity, process admission and wire checks run.
|
|
reading = nts.clock_reading()
|
|
shift = recovery.IO_BUDGET_NS + 1 if case == 'stale-observation' else 0
|
|
sample = nts.Observation(None, recovery.boot_id(), reading.wall - shift,
|
|
reading.before - shift, reading.after - reading.before,
|
|
1_000, reading.after - shift)
|
|
platform = recovery.PlatformQualification(os.uname().release, recovery.clocksource(),
|
|
hashlib.sha256(b'SYNTHETIC C01 test profile; not suspend qualification').hexdigest(),
|
|
50_000_000, 50_000_000, 10)
|
|
with patch.object(nts, 'collect', return_value=sample):
|
|
approval = recovery.prepare(object(), platform, BINDING, bytes([7] * 16))
|
|
authority_pid, authority_control = spawn(authority, keep=(key_fd,))
|
|
assert receive(authority_control) == {'listening': True}
|
|
app_pid, app_control = spawn(application)
|
|
|
|
def administrator(control):
|
|
drop(AUTHORITY)
|
|
channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
channel.settimeout(5); channel.connect(address)
|
|
peer = recovery.Peer(authority_pid + (100000 if case == 'wrong-peer' else 0), AUTHORITY, AUTHORITY)
|
|
try:
|
|
approval.send_once(channel, peer)
|
|
except nts.Unavailable as error:
|
|
assert case != 'accepted-once', str(error)
|
|
expected = 'peer mismatch' if case == 'wrong-peer' else 'stale'
|
|
assert expected in str(error), str(error)
|
|
assert approval._used
|
|
control.send(json.dumps({'refused': expected, 'consumed': True}).encode())
|
|
else:
|
|
assert case == 'accepted-once'
|
|
control.send(b'{"sent":true}')
|
|
|
|
admin_pid, admin_control = spawn(administrator)
|
|
peers = receive(authority_control)
|
|
assert peers == {'app': [app_pid, APP, APP], 'admin': [admin_pid, AUTHORITY, AUTHORITY]}
|
|
authority_control.send(json.dumps({'version': 3, 'role': 'authority', 'process_id': authority_pid,
|
|
'data_dir': str(data), 'store_dir': str(store), 'binding': BINDING,
|
|
'key_file': {'device': key_metadata.st_dev, 'inode': key_metadata.st_ino},
|
|
'peer': {'pid': app_pid, 'uid': APP, 'gid': APP},
|
|
'admin': {'pid': admin_pid, 'uid': AUTHORITY, 'gid': AUTHORITY},
|
|
'provision': True, 'old_writers_excluded': True}).encode())
|
|
result = receive(admin_control)
|
|
finish(admin_pid, 0)
|
|
if case == 'accepted-once':
|
|
assert result == {'sent': True}
|
|
app_control.send(b'{"exercise":true}')
|
|
assert receive(app_control) == {'accepted_once': True, 'replay_refused': True}
|
|
finish(app_pid, 0)
|
|
finish(authority_pid, 1) # Deliberate second Recover freezes service.
|
|
else:
|
|
assert result['consumed']
|
|
finish(authority_pid, 1)
|
|
app_control.send(b'{"exercise":false}')
|
|
finish(app_pid, 0)
|
|
print('PASS C01 recovery handoff:', case)
|
|
finally:
|
|
os.close(key_fd)
|
|
for control in controls:
|
|
control.close()
|
|
for pid in children:
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
for pid in children:
|
|
os.waitpid(pid, 0)
|