252 lines
11 KiB
Python
252 lines
11 KiB
Python
"""Supervised single-slot network mutation owner. Never imports runtime code."""
|
|
import array
|
|
import json
|
|
import os
|
|
import select
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import threading
|
|
import time
|
|
import c01_network_settings_nm as nm
|
|
|
|
PROGRESS = struct.Struct('!IIQQ16s')
|
|
|
|
|
|
def now():
|
|
return time.clock_gettime_ns(time.CLOCK_BOOTTIME)
|
|
|
|
|
|
def exact(pairs):
|
|
result = {}
|
|
for key, value in pairs:
|
|
nm.require(key not in result)
|
|
result[key] = value
|
|
return result
|
|
|
|
|
|
def validate_ticket(ticket):
|
|
nm.require(type(ticket) is dict and set(ticket) == {'epoch', 'sequence'})
|
|
nm.require(type(ticket['sequence']) is int and 0 < ticket['sequence'] < 2**64)
|
|
epoch = ticket['epoch']
|
|
nm.require(type(epoch) is list and len(epoch) == 16 and any(epoch) and
|
|
all(type(byte) is int and 0 <= byte <= 255 for byte in epoch))
|
|
|
|
|
|
def receive(channel, identity, limit=4096):
|
|
raw, anc, flags, _ = channel.recvmsg(limit + 1, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64), socket.MSG_CMSG_CLOEXEC)
|
|
credentials, unexpected = [], False
|
|
for level, kind, data in anc:
|
|
if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
|
|
rights = array.array('i')
|
|
rights.frombytes(data[:len(data) - len(data) % rights.itemsize])
|
|
for fd in rights:
|
|
os.close(fd)
|
|
unexpected = True
|
|
elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS):
|
|
credentials.append(struct.unpack('3i', data))
|
|
else:
|
|
unexpected = True
|
|
nm.require(not unexpected and credentials == [identity] and flags & ~socket.MSG_CMSG_CLOEXEC == 0 and 0 < len(raw) <= limit)
|
|
return json.loads(raw, object_pairs_hook=exact)
|
|
|
|
|
|
def send(channel, value):
|
|
raw = json.dumps(value, separators=(',', ':')).encode()
|
|
nm.require(len(raw) <= 4096 and channel.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw))
|
|
|
|
|
|
def open_journal():
|
|
# Fixed deployment-owned location, no caller path, no aliases in ancestry.
|
|
parent = os.open('/', os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
|
|
try:
|
|
for component in ('var', 'lib', 'archipelago-network-settings'):
|
|
if component == 'archipelago-network-settings':
|
|
try:
|
|
os.mkdir(component, 0o700, dir_fd=parent)
|
|
os.fsync(parent)
|
|
except FileExistsError:
|
|
pass
|
|
child = os.open(component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=parent)
|
|
meta = os.fstat(child)
|
|
if meta.st_uid != 0 or meta.st_gid != 0 or meta.st_mode & 0o022:
|
|
os.close(child)
|
|
raise ValueError('unprotected network journal')
|
|
os.close(parent)
|
|
parent = child
|
|
return parent
|
|
except BaseException:
|
|
os.close(parent)
|
|
raise
|
|
|
|
|
|
class Journal:
|
|
def __init__(self, fd):
|
|
self.fd = fd
|
|
meta = os.fstat(fd)
|
|
nm.require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and meta.st_gid == 0 and not meta.st_mode & 0o077)
|
|
self.record = None
|
|
try:
|
|
file = os.open('operation.json', os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=fd)
|
|
except FileNotFoundError:
|
|
return
|
|
try:
|
|
meta = os.fstat(file)
|
|
nm.require(stat.S_ISREG(meta.st_mode) and meta.st_nlink == 1 and meta.st_uid == 0 and not meta.st_mode & 0o077 and meta.st_size <= 4096)
|
|
self.record = json.loads(os.read(file, 4097), object_pairs_hook=exact)
|
|
nm.require(type(self.record) is dict and set(self.record) == {'ticket', 'phase', 'targets'})
|
|
nm.require(self.record['phase'] in ('Applying', 'Verifying', 'Unknown', 'Succeeded'))
|
|
validate_ticket(self.record['ticket'])
|
|
finally:
|
|
os.close(file)
|
|
|
|
def blocked(self):
|
|
return self.record is not None and self.record['phase'] != 'Succeeded'
|
|
|
|
def save(self, ticket, phase, targets):
|
|
value = {'ticket': ticket, 'phase': phase, 'targets': targets}
|
|
raw = json.dumps(value, separators=(',', ':')).encode()
|
|
nm.require(len(raw) <= 4096)
|
|
name = 'stage-' + os.urandom(16).hex()
|
|
fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600, dir_fd=self.fd)
|
|
try:
|
|
nm.require(os.write(fd, raw) == len(raw))
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
os.rename(name, 'operation.json', src_dir_fd=self.fd, dst_dir_fd=self.fd)
|
|
os.fsync(self.fd)
|
|
self.record = value
|
|
|
|
|
|
class Owner:
|
|
def __init__(self, journal, factory=nm.NetworkManager):
|
|
self.journal, self.factory = journal, factory
|
|
self.epoch = list(os.urandom(16))
|
|
self.counter = 0
|
|
self.receipt = None
|
|
self.end = 0
|
|
self.thread = None
|
|
self.lock = threading.Lock()
|
|
self.revoked = False
|
|
|
|
def submit(self, mutation):
|
|
nm.validate(mutation)
|
|
with self.lock:
|
|
nm.require(not self.revoked)
|
|
nm.require(self.thread is None or not self.thread.is_alive(), 'network slot busy')
|
|
self.counter += 1
|
|
ticket = {'epoch': self.epoch, 'sequence': self.counter}
|
|
self.receipt = {'ticket': ticket, 'phase': 'Rejected' if self.journal.blocked() else 'Accepted'}
|
|
self.end = now() + 45_000_000_000
|
|
if self.receipt['phase'] == 'Accepted':
|
|
self.thread = threading.Thread(target=self.work, args=(mutation, ticket), daemon=True)
|
|
self.thread.start()
|
|
return dict(self.receipt)
|
|
|
|
def work(self, mutation, ticket):
|
|
effect, targets = False, []
|
|
def before(paths):
|
|
nonlocal effect, targets
|
|
with self.lock:
|
|
nm.require(not self.revoked and now() < self.end)
|
|
targets = paths
|
|
# Fence memory before disk: uncertain persistence must freeze too.
|
|
effect = True
|
|
self.journal.save(ticket, 'Applying', targets)
|
|
self.receipt['phase'] = 'Applying'
|
|
def verifying():
|
|
with self.lock:
|
|
nm.require(not self.revoked and now() < self.end)
|
|
self.receipt['phase'] = 'Verifying'
|
|
try:
|
|
self.factory().execute(mutation, self.end / 1e9, before, verifying)
|
|
with self.lock:
|
|
nm.require(not self.revoked and now() < self.end and self.receipt['phase'] == 'Verifying')
|
|
self.journal.save(ticket, 'Succeeded', targets)
|
|
self.receipt['phase'] = 'Succeeded'
|
|
except BaseException:
|
|
with self.lock:
|
|
self.receipt['phase'] = 'Unknown' if effect or self.revoked else 'Rejected'
|
|
if effect:
|
|
self.revoked = True # Cannot clear uncertainty in this lifetime.
|
|
try:
|
|
self.journal.save(ticket, 'Unknown', targets)
|
|
except BaseException:
|
|
pass # Original write-ahead record remains the restart fence.
|
|
finally:
|
|
if 'Wifi' in mutation:
|
|
mutation['Wifi']['password'] = ''
|
|
|
|
def status(self, ticket):
|
|
validate_ticket(ticket)
|
|
with self.lock:
|
|
nm.require(self.receipt is not None and ticket == self.receipt['ticket'])
|
|
if now() >= self.end and self.receipt['phase'] in ('Accepted', 'Applying', 'Verifying'):
|
|
self.revoked = True
|
|
self.receipt['phase'] = 'Unknown'
|
|
return dict(self.receipt)
|
|
|
|
|
|
def main():
|
|
nm.require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0))
|
|
parent = os.getppid()
|
|
nm.require(parent > 1)
|
|
control, app = socket.socket(fileno=3), socket.socket(fileno=4)
|
|
for channel in (control, app):
|
|
nm.require(channel.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) == socket.SOCK_SEQPACKET)
|
|
channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
channel.setblocking(False)
|
|
channel.set_inheritable(False)
|
|
nm.require(struct.unpack('3i', control.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) == (parent, 0, 0))
|
|
nm.require(select.select([control], [], [], 2)[0])
|
|
grant = receive(control, (parent, 0, 0))
|
|
nm.require(set(grant) == {'version', 'process_id', 'application', 'journal_device', 'journal_inode'} and type(grant['version']) is int and grant['version'] == 1 and grant['process_id'] == os.getpid())
|
|
peer = grant['application']
|
|
nm.require(set(peer) == {'pid', 'uid', 'gid'} and all(type(peer[k]) is int and peer[k] > 0 for k in peer))
|
|
identity = (peer['pid'], peer['uid'], peer['gid'])
|
|
nm.require(struct.unpack('3i', app.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) == identity)
|
|
meta = os.fstat(5)
|
|
nm.require((meta.st_dev, meta.st_ino) == (grant['journal_device'], grant['journal_inode']))
|
|
owner = Owner(Journal(5))
|
|
def progress(event, sequence=0, deadline=0):
|
|
raw = PROGRESS.pack(1, event, sequence, deadline, bytes(owner.epoch))
|
|
nm.require(control.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw))
|
|
progress(1)
|
|
exchange, active, heartbeat = 0, False, now()
|
|
while os.getppid() == parent:
|
|
if select.select([control], [], [], 0)[0]:
|
|
raise ValueError('network owner revoked')
|
|
if active:
|
|
state = owner.status(owner.receipt['ticket'])
|
|
nm.require(now() < owner.end, 'network owner deadline')
|
|
if state['phase'] in ('Succeeded', 'Rejected', 'Unknown') and not owner.thread.is_alive():
|
|
progress(4, owner.counter, owner.end)
|
|
active = False
|
|
elif now() >= heartbeat:
|
|
progress(3, owner.counter, owner.end)
|
|
heartbeat = now() + 500_000_000
|
|
elif now() >= heartbeat:
|
|
# Idle liveness also bounds a stall before STARTED can be emitted.
|
|
progress(5)
|
|
heartbeat = now() + 500_000_000
|
|
if not select.select([app], [], [], .05)[0]:
|
|
continue
|
|
request = receive(app, identity)
|
|
nm.require(type(request) is dict and set(request) == {'version', 'sequence', 'operation'} and type(request['sequence']) is int and 0 < request['sequence'] < 2**64 and request['sequence'] == exchange + 1 and type(request['version']) is int and request['version'] == 1)
|
|
operation = request['operation']
|
|
nm.require(type(operation) is dict and len(operation) == 1)
|
|
if 'Submit' in operation:
|
|
result = owner.submit(operation['Submit'])
|
|
if result['phase'] == 'Accepted':
|
|
progress(2, owner.counter, owner.end)
|
|
active = True
|
|
heartbeat = now() + 500_000_000
|
|
elif 'Status' in operation:
|
|
result = owner.status(operation['Status'])
|
|
else:
|
|
raise ValueError('unknown network request')
|
|
exchange += 1
|
|
send(app, {'version': 1, 'sequence': exchange, 'receipt': result})
|
|
raise ValueError('network parent lost')
|