1095 lines
60 KiB
Python
1095 lines
60 KiB
Python
"""Embedded C01 orchestration; native /proc/self/exe remains the service MainPID.
|
|
|
|
Production admission requires live exclusion/runtime/route owners and a measured
|
|
conditional asset/service contract. There is no persisted 'migration done' bit, script selector, test
|
|
environment bypass, automatic provisioning, or replacement of an unreaped role.
|
|
The dependency-ready launch engine below is also exercised in disposable tests.
|
|
"""
|
|
import array
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import pwd
|
|
import select
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import time
|
|
|
|
import c01_children as lifecycle
|
|
import c01_host_watch as host_watch
|
|
import c01_network_settings_owner as network_settings
|
|
import c01_network_settings_watch as network_settings_watch
|
|
import c01_tor
|
|
import c01_radio
|
|
import c01_nts as nts
|
|
import c01_recovery as recovery
|
|
import c01_views as views
|
|
import c01_runtime as runtime_service
|
|
import c01_runtime_view
|
|
import c01_access_client
|
|
import c01_network
|
|
import c01_network_client
|
|
import c01_network_owner
|
|
import c01_d09
|
|
import c01_power
|
|
|
|
# This separate role has its own fixed cgroup budget; HostOwner's limits and
|
|
# two-second operation contract are not repurposed for NM activation.
|
|
lifecycle.ROLES['network-settings'] = (128 * 1024**2, 16)
|
|
|
|
CONFIG = Path('/etc/archipelago-c01/launch.json')
|
|
EPOCH_CONFIG = Path('/etc/archipelago-c01/epoch.json')
|
|
LOCK = Path('/var/lib/archipelago-c01/launch.lock')
|
|
SETUP_NS = 120_000_000_000
|
|
BASE = Path('/run/archipelago-c01')
|
|
LIBC = lifecycle.LIBC
|
|
require = lifecycle.require
|
|
now = host_watch.now
|
|
|
|
|
|
class AdmissionUnavailable(ValueError):
|
|
pass
|
|
|
|
|
|
def close_except(keep):
|
|
for name in os.listdir('/proc/self/fd'):
|
|
fd = int(name)
|
|
if fd > 2 and fd not in keep:
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def root_ancestry(path):
|
|
path = Path(path)
|
|
require(path.is_absolute() and '..' not in path.parts, 'absolute protected path required')
|
|
for parent in path.parents:
|
|
meta = parent.lstat()
|
|
require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022,
|
|
'unprotected input ancestry')
|
|
|
|
|
|
def root_capability(path, size=None):
|
|
"""Validate a storage-key FD WITHOUT reading a byte into the parent heap."""
|
|
root_ancestry(path)
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK)
|
|
try:
|
|
meta = os.fstat(fd)
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and meta.st_nlink == 1 and
|
|
stat.S_IMODE(meta.st_mode) in (0o400, 0o600) and
|
|
(size is None or meta.st_size == size), 'invalid root capability')
|
|
return fd
|
|
except BaseException:
|
|
os.close(fd)
|
|
raise
|
|
|
|
|
|
def key_binding(fd):
|
|
meta = os.fstat(fd)
|
|
require(os.lseek(fd, 0, os.SEEK_CUR) == 0, 'parent consumed a storage key')
|
|
return {'device': meta.st_dev, 'inode': meta.st_ino}
|
|
|
|
|
|
def exact_object(pairs):
|
|
value = {}
|
|
for key, item in pairs:
|
|
require(key not in value, 'duplicate supervisor input')
|
|
value[key] = item
|
|
return value
|
|
|
|
|
|
def root_json(path):
|
|
fd = root_capability(path)
|
|
try:
|
|
before = os.fstat(fd)
|
|
require(0 < before.st_size <= 16384, 'oversized supervisor input')
|
|
raw = os.read(fd, 16385)
|
|
after = os.fstat(fd)
|
|
fields = ('st_dev', 'st_ino', 'st_uid', 'st_gid', 'st_mode', 'st_nlink', 'st_size', 'st_mtime_ns', 'st_ctime_ns')
|
|
require(len(raw) == before.st_size and all(getattr(before, k) == getattr(after, k) for k in fields),
|
|
'supervisor input changed')
|
|
return json.loads(raw, object_pairs_hook=exact_object), hashlib.sha256(raw).hexdigest()
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
class LaunchLock:
|
|
"""Never unlink: one root-owned inode across failures and service restarts."""
|
|
def __init__(self, path=LOCK):
|
|
root_ancestry(path)
|
|
self.fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
|
|
try:
|
|
meta = os.fstat(self.fd)
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and
|
|
stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1,
|
|
'unprotected launch lock')
|
|
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
named = os.stat(path, follow_symlinks=False)
|
|
require((meta.st_dev, meta.st_ino) == (named.st_dev, named.st_ino), 'launch lock replaced')
|
|
except BaseException:
|
|
os.close(self.fd)
|
|
raise
|
|
|
|
def close(self):
|
|
fd, self.fd = self.fd, None
|
|
if fd is not None:
|
|
os.close(fd)
|
|
|
|
def bind_group(self, root):
|
|
"""A crash releases flock, not the obligation to inspect retained tasks.
|
|
|
|
Persist only a kernel ownership locator (never an exclusion attestation).
|
|
A same-boot previous role group blocks even if empty: only its original
|
|
reaper or explicit offline recovery may remove that uncertainty.
|
|
"""
|
|
os.lseek(self.fd, 0, os.SEEK_SET)
|
|
raw = os.read(self.fd, 4097)
|
|
require(len(raw) <= 4096, 'launch ownership locator oversized')
|
|
boot = recovery.boot_id()
|
|
if raw:
|
|
previous = json.loads(raw, object_pairs_hook=exact_object)
|
|
require(set(previous) == {'boot', 'group'}, 'invalid launch ownership locator')
|
|
if previous['boot'] == boot:
|
|
prior = Path(previous['group'])
|
|
require(prior.is_relative_to('/sys/fs/cgroup') and '..' not in prior.parts,
|
|
'invalid prior cgroup locator')
|
|
require(not any((prior / ('c01-' + role)).exists() for role in lifecycle.ROLES),
|
|
'prior protected role retirement remains unresolved')
|
|
for directory in (prior, prior / 'supervisor'):
|
|
if directory.exists():
|
|
tasks = (directory / 'cgroup.procs').read_text().split()
|
|
require(all(int(pid) == os.getpid() for pid in tasks), 'prior supervisor still active')
|
|
require(not any((root / ('c01-' + role)).exists() for role in lifecycle.ROLES),
|
|
'delegated role ownership predates this launch')
|
|
raw = json.dumps({'boot': boot, 'group': str(root)}, separators=(',', ':')).encode()
|
|
os.lseek(self.fd, 0, os.SEEK_SET)
|
|
require(os.write(self.fd, raw) == len(raw), 'launch ownership locator write failed')
|
|
os.ftruncate(self.fd, len(raw))
|
|
os.fsync(self.fd)
|
|
|
|
|
|
def principal(name):
|
|
account = pwd.getpwnam(name)
|
|
require(account.pw_uid > 0 and account.pw_gid > 0 and
|
|
account.pw_shell in ('/usr/sbin/nologin', '/sbin/nologin', '/bin/false') and
|
|
os.getgrouplist(name, account.pw_gid) == [account.pw_gid],
|
|
'managed principal must be non-login and have no supplementary access')
|
|
require(len([p for p in pwd.getpwall() if p.pw_uid == account.pw_uid]) == 1,
|
|
'managed UID has another account alias')
|
|
for path in Path('/proc').iterdir():
|
|
if not path.name.isdecimal():
|
|
continue
|
|
try:
|
|
status = dict(line.split(':', 1) for line in (path / 'status').read_text().splitlines() if ':' in line)
|
|
except FileNotFoundError:
|
|
continue
|
|
require(account.pw_uid not in map(int, status['Uid'].split()), 'managed principal already active')
|
|
return account
|
|
|
|
|
|
def private_identity(directory, uid):
|
|
"""Descriptor-relative no-follow checks; no secret content enters Python."""
|
|
identity = os.open('identity', os.O_PATH | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC,
|
|
dir_fd=directory)
|
|
try:
|
|
meta = os.fstat(identity)
|
|
require(meta.st_uid == uid and stat.S_IMODE(meta.st_mode) == 0o700,
|
|
'private identity directory owner/mode mismatch')
|
|
secret = os.open('nostr_secret', os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=identity)
|
|
try:
|
|
meta = os.fstat(secret)
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_uid == uid and meta.st_nlink == 1 and
|
|
stat.S_IMODE(meta.st_mode) == 0o600 and 0 < meta.st_size <= 256,
|
|
'private identity secret/alias mismatch')
|
|
return meta.st_dev, meta.st_ino
|
|
finally:
|
|
os.close(secret)
|
|
finally:
|
|
os.close(identity)
|
|
|
|
|
|
def load_configuration():
|
|
value, _ = root_json(CONFIG)
|
|
fields = {'version', 'application_user', 'authority_user', 'manager_backing', 'snapshot_backing',
|
|
'authority_backing', 'runtime_data_dir', 'runtime_uid', 'runtime_gid', 'snapshot_key',
|
|
'authority_key', 'binding', 'platform_profile', 'platform_profile_sha256', 'clock_domain',
|
|
'host_ip', 'proxy_path', 'native_sha256'}
|
|
if value.get('version') in (4,5):
|
|
fields.update(('epoch_sha256','power_sha256'))
|
|
if value.get('version') == 5:fields.add('admission_sha256')
|
|
require(type(value) is dict and set(value) == fields and type(value['version']) is int and
|
|
value['version'] in (3, 4, 5), 'unsupported supervisor configuration')
|
|
recovery.binding_bytes(value['binding'])
|
|
require(type(value['clock_domain']) is str and len(value['clock_domain']) == 32 and
|
|
all(c in '0123456789abcdef' for c in value['clock_domain']) and
|
|
value['clock_domain'] != '0' * 32, 'qualified clock domain required')
|
|
app, authority = principal(value['application_user']), principal(value['authority_user'])
|
|
require(len({app.pw_uid, authority.pw_uid, value['runtime_uid'], 0}) == 4 and
|
|
len({app.pw_gid, authority.pw_gid, value['runtime_gid'], 0}) == 4,
|
|
'manager/runtime/authority credentials overlap')
|
|
profile, digest = root_json(value['platform_profile'])
|
|
require(digest == value['platform_profile_sha256'], 'platform qualification digest mismatch')
|
|
qualification = recovery.PlatformQualification(**profile)
|
|
# Structural/profile byte binding precedes owner acquisition. Timing
|
|
# qualification remains mandatory immediately before admission, but must
|
|
# not stand in for measuring the independent exclusion/runtime/route graph.
|
|
return value, app, authority, qualification
|
|
|
|
|
|
def load_epoch_configuration(value):
|
|
require(value['version'] in (4,5), 'epoch-bound launch configuration required')
|
|
epoch, digest = root_json(EPOCH_CONFIG)
|
|
require(digest == value['epoch_sha256'] and set(epoch) == {'version','backing','key','delegation'} and
|
|
type(epoch['version']) is int and epoch['version'] == 1, 'epoch configuration binding')
|
|
require(Path(epoch['backing']).parent == Path(value['authority_backing']).parent and
|
|
epoch['backing'] != value['authority_backing'], 'epoch must share exclusive authority custody')
|
|
delegation = epoch['delegation']
|
|
require({'id','allow_wake','max_awake'} <= set(delegation) <=
|
|
{'id','allow_wake','max_awake','characterized','allow_refresh'}, 'epoch delegation fields')
|
|
recovery.EpochIssuerDelegation(bytes(delegation['id']), delegation['allow_wake'], delegation.get('allow_refresh', False)).validate()
|
|
if delegation.get('characterized') is not None:
|
|
recovery.validate_characterized_model(delegation['characterized'])
|
|
maximum = delegation['max_awake']
|
|
require(set(maximum) == {'secs','nanos'} and type(maximum['secs']) is int and
|
|
0 <= maximum['secs'] < 2**64 and type(maximum['nanos']) is int and
|
|
0 <= maximum['nanos'] < 1_000_000_000, 'epoch awake duration encoding')
|
|
require(maximum['secs'] or maximum['nanos'], 'epoch awake appointment missing')
|
|
return epoch
|
|
|
|
|
|
def admit(value, app, authority, owner=None):
|
|
"""Require live exclusion/runtime/routes and a measured conditional contract.
|
|
|
|
Account flags and persisted readiness booleans cannot replace these owners.
|
|
"""
|
|
if owner is None or not isinstance(owner.access_lease, c01_access_client.AccessLease) or not isinstance(
|
|
owner.runtime_lease, c01_runtime_view.RuntimeLease):
|
|
raise AdmissionUnavailable('C01_ADMISSION_UNAVAILABLE: live access/runtime owners required')
|
|
owner.access_lease.check()
|
|
owner.runtime_lease.check()
|
|
require(isinstance(owner.network_lease, c01_network_client.NetworkLease), 'live network owner required')
|
|
owner.network_lease.check()
|
|
c01_network.require_private(os.getpid(), 'managed')
|
|
import c01_admission
|
|
require(value['version']==5 and isinstance(value.get('admission_sha256'),str) and
|
|
len(value['admission_sha256'])==64 and all(c in '0123456789abcdef' for c in value['admission_sha256']),
|
|
'protected conditional admission appointment required')
|
|
contract,contract_digest=root_json(c01_admission.POLICIES/(value['admission_sha256']+'.json'))
|
|
require(contract_digest==value['admission_sha256'],'admission policy bytes')
|
|
c01_admission.validate(contract,value)
|
|
owner.required_routes=contract['required_routes']
|
|
policy=c01_network_owner.Policy()
|
|
try:
|
|
require(policy.value['version'] in (2,3,4),'owned runtime egress required')
|
|
management=policy.routes.get('management')
|
|
require(management is not None and management['source']=='host' and management['target']=='managed' and
|
|
management['port']==5678 and management['target_uid']==app.pw_uid,
|
|
'required management route appointment')
|
|
require(all(name in policy.routes for name in owner.required_routes),'required route absent from policy')
|
|
prestart_routes=[name for name in owner.required_routes if policy.routes[name]['target']!='managed']
|
|
policy.check()
|
|
finally:policy.close()
|
|
owner.network_lease.require_targets(prestart_routes)
|
|
# A separate, bounded root role measures the policy/assets and actual
|
|
# service dispositions. Its authenticated progress is held through READY
|
|
# and all normal/clock-only operation. No stored flag grants admission.
|
|
owner.admission_hold=c01_admission.Hold(owner,value,close_except,packet)
|
|
owner.admission_hold.check()
|
|
print('C01_CONDITIONAL_ADMISSION_HELD policy='+value['admission_sha256'],flush=True)
|
|
|
|
|
|
def acquire_admission_leases(owner, value, app, authority):
|
|
"""Acquire and retain real capabilities before considering native launch.
|
|
|
|
Partial acquisition is owned by Supervisor.retire, including all failure
|
|
paths; never return a detached success boolean or discard a live lease.
|
|
"""
|
|
require(owner.access_lease is None and owner.runtime_lease is None and owner.network_lease is None,
|
|
'admission capability already acquired')
|
|
c01_network.require_private(os.getpid(), 'managed')
|
|
same_value, launch_digest = root_json(CONFIG)
|
|
require(same_value == value, 'launch configuration changed during admission')
|
|
owner.access_lease = c01_access_client.AccessLease(launch_digest)
|
|
runtime, runtime_digest = runtime_service.load_config()
|
|
require((runtime['manager_uid'], runtime['manager_gid']) == (app.pw_uid, app.pw_gid) and
|
|
(runtime['authority_uid'], runtime['authority_gid']) == (authority.pw_uid, authority.pw_gid) and
|
|
(runtime['runtime_uid'], runtime['runtime_gid']) == (value['runtime_uid'], value['runtime_gid']) and
|
|
runtime['runtime_data_dir'] == value['runtime_data_dir'],
|
|
'runtime service differs from admitted principals/data root')
|
|
require({str(Path(value[name + '_backing']).parent) for name in ('manager', 'snapshot', 'authority')}
|
|
<= set(runtime['private_parents']), 'runtime view omits private backing exclusions')
|
|
owner.runtime_parameters = (app.pw_uid, app.pw_gid, value['runtime_uid'], value['runtime_gid'], runtime_digest)
|
|
owner.runtime_workspace = runtime_workspace_environment(runtime)
|
|
owner.runtime_lease = c01_runtime_view.RuntimeLease(*owner.runtime_parameters)
|
|
owner.network_lease = c01_network_client.NetworkLease(app.pw_uid, value['runtime_uid'])
|
|
require(root_json(CONFIG)[0] == value and runtime_service.load_config() == (runtime, runtime_digest),
|
|
'configuration changed while acquiring lifetime capabilities')
|
|
owner.access_lease.check()
|
|
owner.runtime_lease.check()
|
|
|
|
|
|
def peer(channel):
|
|
return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
|
|
|
|
def packet(channel, payload):
|
|
if not isinstance(payload, bytes):
|
|
payload = json.dumps(payload, separators=(',', ':')).encode()
|
|
require(0 < len(payload) <= 8192, 'supervisor packet bound')
|
|
require(channel.send(payload, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(payload),
|
|
'supervisor packet send uncertain')
|
|
|
|
|
|
def child_wait(channel, expected, deadline):
|
|
while now() < deadline:
|
|
ready, _, _ = select.select([channel], [], [], 0.05)
|
|
if ready:
|
|
raw, ancillary, flags, _ = channel.recvmsg(8193, 0, socket.MSG_DONTWAIT)
|
|
require(raw == expected and not ancillary and not flags and now() < deadline,
|
|
'supervisor setup barrier refused')
|
|
return
|
|
raise ValueError('supervisor setup deadline expired')
|
|
|
|
|
|
def native(executable, role, descriptors, environment):
|
|
copies = {slot: fcntl.fcntl(fd, fcntl.F_DUPFD_CLOEXEC, 32) for slot, fd in descriptors.items()}
|
|
executable = fcntl.fcntl(executable, fcntl.F_DUPFD_CLOEXEC, 32)
|
|
for slot, fd in copies.items():
|
|
os.dup2(fd, slot, inheritable=True)
|
|
close_except(set(copies) | {executable})
|
|
os.execve(executable, ['archipelago', role], environment)
|
|
|
|
|
|
def collect_and_authorize(control, qualification, binding, domain, authority_peer, address, deadline, epoch=None, context=None):
|
|
"""Keyless root role, including all blocking collector/helper work.
|
|
|
|
MainPID owns this entire cgroup BEFORE execution. A stuck chronyc spawn/read/
|
|
reap cannot block MainPID: it kills this role and retains the leader plus all
|
|
descendants until the qualified Children owner actually observes retirement.
|
|
PreparedRecovery never crosses IPC: its actual root collector parent forks
|
|
the permanently dropped administrator only after collecting the observation.
|
|
"""
|
|
require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'collector requires root')
|
|
gate_parent, gate_child = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
try:
|
|
packet(control, b'COLLECTING')
|
|
policy = nts.Policy(frozenset({'time.cloudflare.com', 'nts.netnod.se'}))
|
|
delegation = None
|
|
managed=os.open('/proc/self/ns/net',os.O_RDONLY|os.O_CLOEXEC)
|
|
expected=c01_network.identity(managed)
|
|
try:
|
|
# Collect in the appointed NTS producer realm rather than relying
|
|
# on cross-namespace datagram reply routing. Only this already-owned
|
|
# keyless role joins it; application/authority never inherit it.
|
|
c01_network.enter_host_worker(LIBC)
|
|
if epoch is not None:
|
|
delegation = recovery.EpochIssuerDelegation(bytes(epoch['delegation']['id']), epoch['delegation']['allow_wake'], epoch['delegation'].get('allow_refresh', False))
|
|
maximum = epoch['delegation']['max_awake']
|
|
approval = recovery.prepare_epoch(policy, qualification, binding, context, delegation,
|
|
maximum['secs'] * 1_000_000_000 + maximum['nanos'], deadline)
|
|
else:
|
|
approval = recovery.prepare(policy, qualification, binding, domain)
|
|
finally:
|
|
restored=LIBC.setns(managed,c01_network.CLONE_NEWNET)==0
|
|
os.close(managed)
|
|
require(restored and c01_network.require_private(os.getpid(),'managed')==expected,
|
|
'collector must return to the appointed authority realm')
|
|
parent = os.getpid()
|
|
previous_mask = signal.pthread_sigmask(signal.SIG_BLOCK, lifecycle.STOP_SIGNALS)
|
|
try:
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
try:
|
|
require(LIBC.prctl(1, signal.SIGKILL, 0, 0, 0) == 0 and os.getppid() == parent,
|
|
'collector administrator parent changed')
|
|
close_except({gate_child.fileno()})
|
|
signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
|
|
lifecycle.drop_credentials(authority_peer['uid'], authority_peer['gid'], parent)
|
|
child_wait(gate_child, b'AUTHORIZE', deadline)
|
|
gate_child.close()
|
|
channel = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
channel.connect(address)
|
|
if delegation is not None:
|
|
approval.send_epoch_once(channel, recovery.Peer(**authority_peer), delegation)
|
|
else:
|
|
approval.send_once(channel, recovery.Peer(**authority_peer))
|
|
os._exit(0)
|
|
except BaseException:
|
|
os._exit(111)
|
|
# Register the direct child before any fallible reporting. The root
|
|
# owner already owns its inherited cgroup even if this process dies.
|
|
admin_pid = pid
|
|
finally:
|
|
signal.pthread_sigmask(signal.SIG_SETMASK, previous_mask)
|
|
gate_child.close()
|
|
packet(control, {'pid': admin_pid, 'uid': authority_peer['uid'], 'gid': authority_peer['gid']})
|
|
child_wait(control, b'RELAYED', deadline)
|
|
packet(gate_parent, b'AUTHORIZE')
|
|
gate_parent.close()
|
|
while now() < deadline:
|
|
reaped, status = os.waitpid(admin_pid, os.WNOHANG)
|
|
if reaped:
|
|
require(reaped == admin_pid and os.waitstatus_to_exitcode(status) == 0,
|
|
'recovery administrator failed')
|
|
packet(control, b'ADMIN_SENT')
|
|
# Do not exit/close a queued status before MainPID consumes it.
|
|
# Authority acceptance is independently required by MainPID.
|
|
child_wait(control, b'RELEASE', deadline)
|
|
return
|
|
time.sleep(0.01)
|
|
raise ValueError('collector administrator setup expired')
|
|
finally:
|
|
gate_parent.close()
|
|
gate_child.close()
|
|
# No blocking cleanup wait: any failure exits the leader; MainPID owns
|
|
# the still-unreaped descendant set, including uninterruptible helpers.
|
|
|
|
|
|
def start_collector(owner, qualification, binding, domain, authority_peer, address, deadline, epoch=None, context=None):
|
|
# No storage-key descriptors or other parent's path capabilities are kept.
|
|
# Call only after authenticated STORE_READY; prepare() is never called by
|
|
# the root observer itself, including validation and helper construction.
|
|
return owner.children.spawn('collector', lambda control: collect_and_authorize(
|
|
control, qualification, binding, domain, authority_peer, address, deadline, epoch, context))
|
|
|
|
|
|
class Supervisor:
|
|
"""One owner through RUNNING -> STOPPING -> reaped, with no restart branch."""
|
|
def __init__(self, lock):
|
|
self.lock = lock
|
|
self.children = lifecycle.Children()
|
|
try:
|
|
lock.bind_group(self.children.root)
|
|
except BaseException:
|
|
self.children.close()
|
|
raise
|
|
self.watch = None
|
|
self.network_settings_watch = None
|
|
self.tor_watch = None
|
|
self.runtime_lease = None
|
|
self.access_lease = None
|
|
self.network_lease = None
|
|
self.runtime_parameters = None
|
|
self.runtime_workspace = None
|
|
self.failure = None
|
|
self.completed_one_shots = {'administrator'}
|
|
self.notifier = None
|
|
self.power_lease = None
|
|
self.power_channel = None
|
|
self.power_state = None
|
|
self.power_epoch_revoked = False
|
|
self.authority_record = None
|
|
self.authority_credentials = None
|
|
self.power_command = b'SLEEP_REVOKE'
|
|
self.clock_sequence = 0
|
|
self.prior_power_signal = signal.signal(signal.SIGUSR1, lambda *_: None)
|
|
self.wake_read, self.wake_write = os.pipe2(os.O_NONBLOCK | os.O_CLOEXEC)
|
|
self.prior_wakeup = signal.set_wakeup_fd(self.wake_write)
|
|
|
|
def notify(self, message):
|
|
if self.notifier is not None:
|
|
require(self.notifier.send(message, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(message),
|
|
'service notification failed')
|
|
|
|
def stop(self, reason):
|
|
if self.failure is None:
|
|
try:os.write(2,('C01_SUPERVISOR_STOP reason='+str(reason)+'\n').encode())
|
|
except OSError:pass
|
|
self.failure = self.failure or reason
|
|
self.children.request_stop()
|
|
if self.watch:
|
|
self.watch.revoke(reason)
|
|
if self.network_settings_watch:
|
|
self.network_settings_watch.revoke(reason)
|
|
if self.tor_watch:
|
|
self.tor_watch.revoke(reason)
|
|
|
|
def tick(self):
|
|
hold=getattr(self,'admission_hold',None)
|
|
if hold is not None and not self.children.stopping:
|
|
try:hold.check()
|
|
except (OSError,ValueError) as error:
|
|
try:os.write(2,('C01_ADMISSION_REVOKED type='+type(error).__name__+' reason='+str(error)+'\n').encode())
|
|
except OSError:pass
|
|
self.stop('admission lifetime lost')
|
|
if self.power_lease is not None and not self.children.stopping:
|
|
try:
|
|
request = self.power_lease.request()
|
|
if request is not None and self.power_state is None:
|
|
self.power_state = request
|
|
self.power_channel = self.power_lease.connect(request)
|
|
if self.authority_record is None:
|
|
self.stop('power requested before epoch authority exists')
|
|
else:
|
|
packet(self.authority_record.channel, self.power_command)
|
|
self.power_deadline = now() + 3_000_000_000
|
|
if self.power_state is not None and not self.children.stopping:
|
|
channel = self.authority_record.channel
|
|
if select.select([channel], [], [], 0)[0]:
|
|
raw, anc, flags, _ = channel.recvmsg(8193, socket.CMSG_SPACE(12), socket.MSG_CMSG_CLOEXEC)
|
|
require(flags & ~socket.MSG_CMSG_CLOEXEC == 0 and anc ==
|
|
[(socket.SOL_SOCKET, socket.SCM_CREDENTIALS, struct.pack('3i', *self.authority_credentials))],
|
|
'power epoch acknowledgement credentials')
|
|
self.power_epoch_revoked = raw == b'EPOCH_REVOKED'
|
|
self.stop('power caller retirement')
|
|
elif now() >= self.power_deadline:
|
|
self.stop('power epoch revocation uncertain')
|
|
except (OSError, ValueError):
|
|
self.stop('power authority lost')
|
|
if not self.children.stopping:
|
|
for lease, reason in ((self.access_lease, 'access lease lost'),
|
|
(self.runtime_lease, 'runtime lease lost'),
|
|
(self.network_lease, 'network lease lost')):
|
|
if lease is not None:
|
|
try:
|
|
lease.check()
|
|
except (OSError, ValueError):
|
|
self.stop(reason)
|
|
if self.watch and not self.children.stopping and not self.watch.receive():
|
|
self.stop('host worker failed')
|
|
if self.network_settings_watch and not self.children.stopping and not self.network_settings_watch.receive():
|
|
self.stop('network settings worker failed')
|
|
if self.tor_watch and not self.children.stopping and not self.tor_watch.receive():
|
|
self.stop('Tor lifecycle worker failed')
|
|
errors = self.children.poll()
|
|
if errors:
|
|
self.stop('child retirement error')
|
|
for role, code in self.children.results.items():
|
|
if role not in self.completed_one_shots or code != 0:
|
|
self.stop('supervised role exited')
|
|
if self.children.stopping and self.watch:
|
|
self.watch.revoke('supervisor stopping')
|
|
if self.children.stopping and self.network_settings_watch:
|
|
self.network_settings_watch.revoke('supervisor stopping')
|
|
if self.children.stopping and self.tor_watch:
|
|
self.tor_watch.revoke('supervisor stopping')
|
|
if self.watch and 'host-io' in self.children.results:
|
|
self.watch.release_reaped()
|
|
if self.network_settings_watch and 'network-settings' in self.children.results:
|
|
self.network_settings_watch.release_reaped()
|
|
if self.tor_watch and 'tor-owner' in self.children.results:
|
|
self.tor_watch.release_reaped()
|
|
return not self.children.stopping
|
|
|
|
def receive(self, record, uid, gid, deadline):
|
|
while now() < deadline and self.tick():
|
|
require(record.role in self.children.children and record.pid is not None and
|
|
record.channel.fileno() >= 0, 'child exited before setup status')
|
|
ready, _, _ = select.select([record.channel, self.wake_read], [], [], 0.05)
|
|
if self.wake_read in ready:
|
|
os.read(self.wake_read, 4096)
|
|
if record.channel in ready:
|
|
raw, anc, flags, _ = record.channel.recvmsg(8193, socket.CMSG_SPACE(12), socket.MSG_CMSG_CLOEXEC)
|
|
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)
|
|
require(flags & ~socket.MSG_CMSG_CLOEXEC == 0 and anc ==
|
|
[(socket.SOL_SOCKET, socket.SCM_CREDENTIALS, struct.pack('3i', record.pid, uid, gid))] and
|
|
0 < len(raw) <= 8192 and now() < deadline, 'child setup status refused')
|
|
return raw
|
|
raise ValueError('supervisor stopped or setup expired')
|
|
|
|
def wait(self, record, expected, uid, gid, deadline):
|
|
require(self.receive(record, uid, gid, deadline) == expected, 'child setup status refused')
|
|
|
|
def relay_collector(self, collector, authority, uid, gid, deadline):
|
|
self.wait(collector, b'COLLECTING', 0, 0, deadline)
|
|
admin = json.loads(self.receive(collector, 0, 0, deadline), object_pairs_hook=exact_object)
|
|
require(type(admin) is dict and set(admin) == {'pid', 'uid', 'gid'} and
|
|
type(admin['pid']) is int and admin['pid'] > 1 and
|
|
admin['pid'] not in (os.getpid(), collector.pid, authority.pid) and
|
|
type(admin['uid']) is int and type(admin['gid']) is int and
|
|
(admin['uid'], admin['gid']) == (uid, gid), 'collector administrator identity refused')
|
|
# Only exact late peer credentials cross this channel, never time data.
|
|
packet(authority.channel, admin)
|
|
packet(collector.channel, b'RELAYED')
|
|
return admin
|
|
|
|
def retire(self):
|
|
self.stop('retiring')
|
|
# Deliberately no deadline that releases the lock over a D-state worker.
|
|
# Service TimeoutStopSec=infinity and Restart=no preserve this owner.
|
|
while True:
|
|
try:
|
|
self.tick()
|
|
if not self.children.children and not self.children.reserved:
|
|
self.children.close()
|
|
break
|
|
ready, _, _ = select.select([self.wake_read], [], [], 0.05)
|
|
if ready:
|
|
os.read(self.wake_read, 4096)
|
|
except (OSError, ValueError):
|
|
self.children.request_stop()
|
|
time.sleep(0.05)
|
|
signal.set_wakeup_fd(self.prior_wakeup)
|
|
os.close(self.wake_read)
|
|
os.close(self.wake_write)
|
|
if self.runtime_lease is not None:
|
|
self.runtime_lease.close()
|
|
if self.access_lease is not None:
|
|
self.access_lease.close()
|
|
if self.network_lease is not None:
|
|
self.network_lease.close()
|
|
self.lock.close()
|
|
if self.power_channel is not None:
|
|
try:
|
|
packet(self.power_channel, {'nonce':self.power_state['nonce'],
|
|
'epoch_revoked':self.power_epoch_revoked,'callers_reaped':True})
|
|
except (OSError, ValueError):
|
|
# The power owner retains Draining and cannot permit sleep from
|
|
# a lost reply. All children are already reaped; finish closing
|
|
# our capabilities without manufacturing a second receipt.
|
|
self.failure = 'power retirement receipt delivery uncertain'
|
|
finally: self.power_channel.close()
|
|
if self.power_lease is not None:
|
|
self.power_lease.close()
|
|
signal.signal(signal.SIGUSR1, self.prior_power_signal)
|
|
if self.notifier is not None:
|
|
self.notifier.close()
|
|
|
|
|
|
class Prepared:
|
|
"""Prestage all paths, pinned keys, mounts and payloads before live NTS."""
|
|
def __init__(self, value):
|
|
self.fds = []
|
|
self.value = value
|
|
try:
|
|
self.epoch = load_epoch_configuration(value)
|
|
self.executable = self.own(os.open('/proc/self/exe', os.O_RDONLY | os.O_CLOEXEC))
|
|
meta = os.fstat(self.executable)
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022,
|
|
'native MainPID is not root protected')
|
|
with open(f'/proc/self/fd/{self.executable}', 'rb') as stream:
|
|
require(hashlib.file_digest(stream, 'sha256').hexdigest() == value['native_sha256'],
|
|
'native artifact differs from provisioning')
|
|
self.keys = {role: self.own(root_capability(value[name], 32)) for role, name in
|
|
(('application', 'snapshot_key'), ('authority', 'authority_key'))}
|
|
self.keys['epoch'] = self.own(root_capability(self.epoch['key'], 32))
|
|
require(len({tuple(key_binding(fd).values()) for fd in self.keys.values()}) == 3, 'epoch/authority/application key alias')
|
|
require(key_binding(self.keys['application']) != key_binding(self.keys['authority']),
|
|
'storage keys alias the same inode')
|
|
self.backings = {name: self.own(views.require_private_backing(value[name + '_backing']))
|
|
for name in ('manager', 'snapshot', 'authority')}
|
|
self.backings['epoch'] = self.own(views.require_private_backing(self.epoch['backing']))
|
|
paths = [Path(value[name + '_backing']) for name in ('manager','snapshot','authority')]
|
|
paths.append(Path(self.epoch['backing']))
|
|
runtime = Path(value['runtime_data_dir'])
|
|
paths.append(runtime)
|
|
require(len(set(paths)) == 5 and all(not a.is_relative_to(b) for a in paths for b in paths if a != b),
|
|
'private and runtime roots overlap')
|
|
for name, fd in self.backings.items():
|
|
require(views.identity(fd)[2:] == (value['runtime_uid'], value['runtime_gid'], 0o700),
|
|
'private backing must preserve original runtime ownership and mode')
|
|
private_identity(self.backings['manager'], value['runtime_uid'])
|
|
self.backing_sources = {name: views.PinnedSource.capture(fd, self.epoch['backing'] if name == 'epoch' else value[name + '_backing'])
|
|
for name, fd in self.backings.items()}
|
|
root_ancestry(runtime)
|
|
self.runtime = self.own(os.open(runtime, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC))
|
|
require(runtime.resolve(strict=True) == runtime, 'runtime root must be canonical')
|
|
self.runtime_source = views.PinnedSource.capture(self.runtime, runtime)
|
|
root_ancestry(value['proxy_path'])
|
|
self.proxy = self.own(os.open(value['proxy_path'], os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC))
|
|
meta = os.fstat(self.proxy)
|
|
require(stat.S_ISREG(meta.st_mode) and meta.st_uid == 0 and meta.st_nlink == 1 and
|
|
not meta.st_mode & 0o6022 and meta.st_mode & 0o005 == 0o005,
|
|
'runtime proxy not protected')
|
|
self.proxy_source = views.PinnedSource.capture(self.proxy, value['proxy_path'])
|
|
except BaseException:
|
|
self.close()
|
|
raise
|
|
|
|
def own(self, fd):
|
|
self.fds.append(fd)
|
|
return fd
|
|
|
|
def close(self):
|
|
while self.fds:
|
|
os.close(self.fds.pop())
|
|
|
|
def role_views(self, account, role, runtime_view=None):
|
|
mounts = views.PrivateMounts()
|
|
# Filesystem Unix sockets (user/system buses, Podman, terminal sockets)
|
|
# are absent. Abstract Unix IPC additionally requires admission's private
|
|
# routed network namespace; pathname hiding alone never attests that.
|
|
for target in ('/run', '/home', '/root', '/tmp', '/var/tmp'):
|
|
mounts.cover(target, 0o755 if target == '/run' else 0o700)
|
|
BASE.mkdir(mode=0o755)
|
|
BASE.chmod(0o755) # UMask=0077 must not hide the role's owned views.
|
|
if role == 'authority':
|
|
# The common binding names the manager root, but authority gets no
|
|
# manager capability. A root-owned empty placeholder satisfies the
|
|
# native canonical-path check without exposing manager state.
|
|
(BASE / 'manager').mkdir(mode=0o700)
|
|
chosen = ('manager', 'snapshot') if role == 'application' else ('authority', 'epoch')
|
|
for name in chosen:
|
|
destination = BASE / name
|
|
destination.mkdir(mode=0o700)
|
|
mounts.idmapped_directory(self.backing_sources[name], destination,
|
|
self.value['runtime_uid'], account.pw_uid, self.value['runtime_gid'], account.pw_gid)
|
|
if role == 'application':
|
|
mounts.idmapped_directory(self.runtime_source, self.value['runtime_data_dir'],
|
|
self.value['runtime_uid'], account.pw_uid, self.value['runtime_gid'], account.pw_gid)
|
|
install_proxy_aliases(self.proxy_source)
|
|
if runtime_view is not None:
|
|
runtime_view.install()
|
|
else:
|
|
mounts.cover(self.value['runtime_data_dir'], 0o700)
|
|
# Backing parents remain root-private; do not unhide them for role UIDs.
|
|
|
|
|
|
def install_proxy_aliases(proxy):
|
|
"""Root-created, read-only exact-inode aliases in the private /run mount.
|
|
|
|
PATH is convenience only: native AppArmor/Landlock/seccomp enforce execution.
|
|
Production application setup additionally installs its independently acquired
|
|
runtime socket capability. A view-only fixture without it remains fail-closed.
|
|
"""
|
|
require(isinstance(proxy, views.PinnedSource) and not proxy.directory and
|
|
proxy.namespace != views.namespace_identity(), 'pre-namespace pinned proxy locator required')
|
|
require(not proxy.checked_protections() & os.ST_NOEXEC, 'proxy source is noexec')
|
|
directory = BASE / 'bin'
|
|
directory.mkdir(mode=0o755)
|
|
directory.chmod(0o755)
|
|
target = directory / 'archipelago-runtime-proxy'
|
|
target.touch(mode=0o555, exist_ok=False)
|
|
local = proxy.reopen()
|
|
try:
|
|
require(not os.fstatvfs(local).f_flag & os.ST_NOEXEC, 'proxy locator mount is noexec')
|
|
require(views.LIBC.mount(f'/proc/self/fd/{local}'.encode(), os.fsencode(target), None, 4096, None) == 0,
|
|
'pinned proxy bind failed')
|
|
finally:
|
|
os.close(local)
|
|
require(views.LIBC.mount(None, os.fsencode(target), None, 4096 | 32 | 1 | 2 | 4, None) == 0,
|
|
'read-only proxy bind failed')
|
|
proxy.checked_protections()
|
|
before, after = os.fstat(proxy.fd), target.stat()
|
|
require((before.st_dev, before.st_ino) == (after.st_dev, after.st_ino), 'proxy bind identity mismatch')
|
|
require(os.statvfs(target).f_flag & os.ST_RDONLY and not os.statvfs(target).f_flag & os.ST_NOEXEC,
|
|
'proxy mount protection mismatch')
|
|
for command in ('podman', 'systemctl', 'journalctl', 'sh', 'bash', 'tmux', 'loginctl', 'mkdir',
|
|
'rm', 'cp', 'mv', 'cat', 'test', 'which', 'curl', 'git', 'df', 'du'):
|
|
(directory / command).symlink_to('archipelago-runtime-proxy')
|
|
|
|
|
|
def runtime_workspace_environment(runtime):
|
|
# A pathname in the separately qualified runtime view, never manager HOME.
|
|
# The handler validates/creates it through the proxy in that actual context.
|
|
return {'ARCHIPELAGO_RUNTIME_WORKSPACE': str(Path(runtime['home']) / 'Work')}
|
|
|
|
|
|
def launch(owner, prepared, app, authority, qualification):
|
|
"""Internal engine. Caller must hold enforced admission for its entire life."""
|
|
value = prepared.value
|
|
require(owner.access_lease is not None and owner.runtime_lease is not None and owner.network_lease is not None and
|
|
owner.runtime_parameters is not None and owner.runtime_workspace is not None and owner.tick(),
|
|
'live access/runtime capabilities required before launch')
|
|
parent, children = os.getpid(), owner.children
|
|
setup_end = now() + SETUP_NS
|
|
authority_address = '\0archy-c01-authority-' + os.urandom(16).hex()
|
|
host_address = '\0archy-c01-host-' + os.urandom(16).hex()
|
|
listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
listener.bind(host_address)
|
|
listener.listen(1)
|
|
listener.settimeout(1)
|
|
settings_address = '\0archy-c01-network-settings-' + os.urandom(16).hex()
|
|
settings_listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
settings_listener.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
settings_listener.bind(settings_address)
|
|
settings_listener.listen(1)
|
|
settings_listener.settimeout(1)
|
|
journal = network_settings.open_journal()
|
|
tor_listener = c01_tor.listener()
|
|
radio_channel = c01_radio.connect()
|
|
radio_pid = None
|
|
tor_address = tor_listener.getsockname()
|
|
tor_dirs = c01_tor.capabilities(prepared.runtime)
|
|
for fd in tor_dirs:
|
|
prepared.own(fd)
|
|
env = {'PATH': str(BASE / 'bin'), 'HOME': str(BASE / 'manager'), 'LANG': 'C.UTF-8',
|
|
'ARCHIPELAGO_DATA_DIR': str(BASE / 'manager'),
|
|
'ARCHIPELAGO_RUNTIME_DATA_DIR': value['runtime_data_dir'],
|
|
'ARCHIPELAGO_HOST_IP': value['host_ip'], 'ARCHIPELAGO_BIND': '127.0.0.1:5678',
|
|
'ARCHIPELAGO_USE_QUADLET_BACKENDS': 'true'}
|
|
env.update(owner.runtime_workspace)
|
|
if 'INVOCATION_ID' in os.environ:
|
|
env['INVOCATION_ID'] = os.environ['INVOCATION_ID']
|
|
|
|
def authority_child(control):
|
|
prepared.role_views(authority, 'authority')
|
|
keep = {prepared.executable, prepared.keys['authority'], prepared.keys['epoch'], control.fileno()}
|
|
close_except(keep) # In particular, NO application key survives the drop.
|
|
lifecycle.drop_credentials(authority.pw_uid, authority.pw_gid, parent)
|
|
server = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
server.bind(authority_address)
|
|
server.listen(2)
|
|
server.settimeout(120)
|
|
packet(control, b'LISTENING')
|
|
connection, _ = server.accept()
|
|
require(peer(connection)[1:] == (app.pw_uid, app.pw_gid), 'authority app credentials refused')
|
|
packet(control, b'CONNECTED')
|
|
child_wait(control, b'EXEC', setup_end)
|
|
native(prepared.executable, 'c01-authority-v4',
|
|
{3: control.fileno(), 4: connection.fileno(), 5: server.fileno(), 9: prepared.keys['authority'], 13: prepared.keys['epoch']}, env)
|
|
|
|
def application_child(control):
|
|
control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
# Never let parent/child drain a fork-shared renewal channel. Acquire the
|
|
# child's socket-mount capability independently before covering /run.
|
|
runtime_view = c01_runtime_view.RuntimeLease(*owner.runtime_parameters)
|
|
try:
|
|
prepared.role_views(app, 'application', runtime_view)
|
|
finally:
|
|
runtime_view.close()
|
|
close_except({control.fileno(), prepared.executable, prepared.keys['application'], prepared.proxy} |
|
|
({radio_channel.fileno()} if radio_channel is not None else set()))
|
|
lifecycle.drop_credentials(app.pw_uid, app.pw_gid, parent)
|
|
clock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
clock.connect(authority_address)
|
|
host = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
host.connect(host_address)
|
|
settings = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
settings.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
settings.connect(settings_address)
|
|
tor = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
tor.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
tor.connect(tor_address)
|
|
packet(control, b'CONNECTED')
|
|
d09 = c01_d09.receive_child(control, parent, setup_end)
|
|
c01_d09.wait_exec(control, parent, setup_end)
|
|
native(prepared.executable, 'c01-managed-v7' if radio_channel is not None else 'c01-managed-v6', {3: control.fileno(), 4: clock.fileno(),
|
|
6: host.fileno(), 8: prepared.proxy, 9: prepared.keys['application'], 10: d09,
|
|
11: settings.fileno(), 12: tor.fileno(),
|
|
**({14: radio_channel.fileno()} if radio_channel is not None else {})},
|
|
env)
|
|
|
|
try:
|
|
authority_child_record = children.spawn('authority', authority_child, prepared.fds)
|
|
owner.authority_record = authority_child_record
|
|
owner.authority_credentials = (authority_child_record.pid, authority.pw_uid, authority.pw_gid)
|
|
owner.wait(authority_child_record, b'LISTENING', authority.pw_uid, authority.pw_gid, setup_end)
|
|
application = children.spawn('application', application_child,
|
|
c01_radio.application_fds(prepared.fds, radio_channel))
|
|
owner.wait(application, b'CONNECTED', app.pw_uid, app.pw_gid, setup_end)
|
|
if radio_channel is not None:
|
|
radio_pid = c01_radio.bind(radio_channel, application.pid, app.pw_uid, app.pw_gid)
|
|
radio_channel.close()
|
|
d09_session, d09_fd = c01_d09.register(owner, application.pid, setup_end)
|
|
try:
|
|
c01_d09.send_child(application.channel, d09_session, d09_fd)
|
|
finally:
|
|
os.close(d09_fd)
|
|
owner.wait(authority_child_record, b'CONNECTED', authority.pw_uid, authority.pw_gid, setup_end)
|
|
accepted, _ = listener.accept()
|
|
require(peer(accepted) == (application.pid, app.pw_uid, app.pw_gid), 'host app peer mismatch')
|
|
app_peer = {'pid': application.pid, 'uid': app.pw_uid, 'gid': app.pw_gid}
|
|
authority_peer = {'pid': authority_child_record.pid, 'uid': authority.pw_uid, 'gid': authority.pw_gid}
|
|
tor_accepted, _ = tor_listener.accept()
|
|
tor_worker = c01_tor.spawn(owner, prepared.executable, tor_accepted,
|
|
app_peer, tor_dirs, native, packet, env)
|
|
|
|
settings_accepted, _ = settings_listener.accept()
|
|
require(peer(settings_accepted) == (application.pid, app.pw_uid, app.pw_gid), 'network settings app peer mismatch')
|
|
settings_accepted.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
def settings_child(control):
|
|
control.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
|
|
require(LIBC.prctl(38, 1, 0, 0, 0) == 0, 'network settings NNP unavailable')
|
|
c01_network.enter_host_worker(LIBC)
|
|
packet(control, b'NETWORK_CHANNEL')
|
|
native(prepared.executable, 'network-settings-owner-v1',
|
|
{3: control.fileno(), 4: settings_accepted.fileno(), 5: journal}, env)
|
|
settings_worker = children.spawn('network-settings', settings_child,
|
|
(settings_accepted.fileno(), journal, prepared.executable))
|
|
owner.wait(settings_worker, b'NETWORK_CHANNEL', 0, 0, setup_end)
|
|
owner.network_settings_watch = network_settings_watch.Watch(settings_worker.channel, settings_accepted, settings_worker.pid)
|
|
journal_meta = os.fstat(journal)
|
|
packet(settings_worker.channel, {'version': 1, 'process_id': settings_worker.pid, 'application': app_peer,
|
|
'journal_device': journal_meta.st_dev, 'journal_inode': journal_meta.st_ino})
|
|
|
|
def host_child(control):
|
|
require(LIBC.prctl(38, 1, 0, 0, 0) == 0, 'host NNP unavailable')
|
|
c01_network.enter_host_worker(LIBC)
|
|
native(prepared.executable, 'host-access-owner',
|
|
{3: control.fileno(), 4: accepted.fileno(), 5: prepared.runtime}, env)
|
|
|
|
worker = children.spawn('host-io', host_child, (accepted.fileno(), prepared.runtime, prepared.executable))
|
|
owner.watch = host_watch.HostWatch(worker.channel, accepted, worker.pid, now() + host_watch.SETUP_LIMIT_NS)
|
|
runtime_meta = os.fstat(prepared.runtime)
|
|
packet(worker.channel, {'version': 2, 'process_id': worker.pid, 'application': app_peer,
|
|
'data_device': runtime_meta.st_dev, 'data_inode': runtime_meta.st_ino})
|
|
# Validate the actual finite role handshakes before releasing application
|
|
# execution. Clock authorization below still requires a fresh barrier.
|
|
for watch,label in ((owner.watch,'host'),(owner.network_settings_watch,'network'),(owner.tor_watch,'Tor')):
|
|
while watch.state=='starting' and now()<setup_end and owner.tick():time.sleep(.01)
|
|
require(watch.state in ('idle','busy') and owner.tick(),label+' owner readiness unavailable')
|
|
require(getattr(owner,'admission_hold',None) is not None,'missing live admission checker')
|
|
# Only the enforced-admission caller may reach here. main never does so
|
|
# while admit() reports unavailable; no stored boolean is consulted.
|
|
common = {'version': 3, 'data_dir': str(BASE / 'manager'), 'binding': value['binding'],
|
|
'provision': False, 'old_writers_excluded': True, 'staged_recovery': True}
|
|
packet(authority_child_record.channel, b'EXEC')
|
|
packet(authority_child_record.channel, {**common, 'role': 'authority', 'process_id': authority_child_record.pid,
|
|
'store_dir': str(BASE / 'authority'), 'key_file': key_binding(prepared.keys['authority']),
|
|
'peer': app_peer, 'admin': None, 'version':4,
|
|
'epoch':{'store_dir':str(BASE/'epoch'),'key_file':key_binding(prepared.keys['epoch']),
|
|
'delegation':prepared.epoch['delegation']}})
|
|
owner.wait(authority_child_record, b'STORE_READY', authority.pw_uid, authority.pw_gid, setup_end)
|
|
epoch_context = json.loads(owner.receive(authority_child_record, authority.pw_uid, authority.pw_gid, setup_end), object_pairs_hook=exact_object)
|
|
require(set(epoch_context) == {'status','challenge'} and epoch_context['status'] == 'EPOCH_CONTEXT',
|
|
'durable epoch context missing')
|
|
# Blocking NTS/helper work runs only inside this already-owned, keyless
|
|
# root role. MainPID keeps ticking all earlier roles while it is pending.
|
|
collector = start_collector(owner, qualification, value['binding'], bytes.fromhex(value['clock_domain']),
|
|
authority_peer, authority_address, setup_end, prepared.epoch, epoch_context['challenge'])
|
|
owner.relay_collector(collector, authority_child_record, authority.pw_uid, authority.pw_gid, setup_end)
|
|
owner.wait(authority_child_record, b'AUTHORIZED', authority.pw_uid, authority.pw_gid, setup_end)
|
|
owner.wait(collector, b'ADMIN_SENT', 0, 0, setup_end)
|
|
owner.completed_one_shots.add('collector')
|
|
packet(collector.channel, b'RELEASE')
|
|
packet(application.channel, b'EXEC')
|
|
packet(application.channel, {**common, 'role': 'application', 'process_id': application.pid,
|
|
'version': 7 if radio_pid is not None else 6, 'radio_owner_pid': radio_pid,
|
|
'd09_session': d09_session, 'network_settings_pid': settings_worker.pid,
|
|
'tor_owner_pid': tor_worker.pid,
|
|
'runtime_data_dir': value['runtime_data_dir'], 'store_dir': str(BASE / 'snapshot'),
|
|
'key_file': key_binding(prepared.keys['application']), 'peer': authority_peer, 'admin': None,
|
|
'clock_refresh': {'version':1,'address':authority_address[1:]} if prepared.epoch['delegation'].get('allow_refresh',False) else None})
|
|
while True:
|
|
status = owner.receive(application, app.pw_uid, app.pw_gid, setup_end)
|
|
require(status != b'FAILED', 'application protected recovery failed; see native startup diagnostic')
|
|
ready = json.loads(status, object_pairs_hook=exact_object)
|
|
if ready.get('status') != 'CLOCK_REFRESH':
|
|
break
|
|
clock_refresh(owner,prepared,qualification,application,authority_child_record,app_peer,authority_peer,authority_address,ready)
|
|
require(type(ready) is dict and set(ready) == {'status', 'listener_fd'} and
|
|
ready['status'] == 'APP_READY' and type(ready['listener_fd']) is int and
|
|
0 <= ready['listener_fd'] <= 65535, 'application listener registration status')
|
|
owner.network_lease.register_management(application.pid, ready['listener_fd'])
|
|
if owner.network_lease.policy.value['version']==4:
|
|
# Managed gate/peer/TLS listeners cannot exist before the native
|
|
# process starts. Wait for their independent root registration,
|
|
# keeping the existing absolute setup deadline and every live lease.
|
|
while True:
|
|
require(owner.tick() and now()<setup_end,'post-start network admission expired')
|
|
owner.network_lease.register_managed_listeners(application.pid,owner.required_routes)
|
|
if all(owner.network_lease.request({'method':'inspect','name':name}).get('producer') is not None
|
|
for name in owner.required_routes):
|
|
break
|
|
time.sleep(.02)
|
|
owner.network_lease.require_targets(owner.required_routes)
|
|
# APP_READY requires native protected recovery AND a bound listener.
|
|
# Required role progress and the independent admission hold must still
|
|
# be live when native recovery and listener registration have completed.
|
|
while owner.watch.state == 'starting' and now() < setup_end and owner.tick():
|
|
time.sleep(0.01)
|
|
require(owner.watch.state in ('idle', 'busy') and owner.tick(), 'host readiness unavailable')
|
|
while owner.network_settings_watch.state == 'starting' and now() < setup_end and owner.tick():
|
|
time.sleep(0.01)
|
|
require(owner.network_settings_watch.state in ('idle', 'busy') and owner.tick(), 'network settings readiness unavailable')
|
|
while owner.tor_watch.state == 'starting' and now() < setup_end and owner.tick():
|
|
time.sleep(0.01)
|
|
require(owner.tor_watch.state in ('idle', 'busy') and owner.tick(), 'Tor lifecycle readiness unavailable')
|
|
require({'application','authority','host-io','network-settings','tor-owner','admission'} <= set(owner.children.children)
|
|
and 'collector' in owner.completed_one_shots and owner.tick(), 'required role lifetimes missing')
|
|
print('C01_BASELINE_READY application_pid='+str(application.pid),flush=True)
|
|
owner.notify(b'READY=1\nSTATUS=C01 protected application recovered and listening')
|
|
while owner.tick():
|
|
ready, _, _ = select.select([owner.wake_read,application.channel], [], [], 0.05)
|
|
if owner.wake_read in ready:
|
|
os.read(owner.wake_read, 4096)
|
|
if application.channel in ready:
|
|
request=json.loads(owner.receive(application,app.pw_uid,app.pw_gid,now()+SETUP_NS),object_pairs_hook=exact_object)
|
|
clock_refresh(owner,prepared,qualification,application,authority_child_record,app_peer,authority_peer,authority_address,request)
|
|
finally:
|
|
listener.close()
|
|
settings_listener.close()
|
|
tor_listener.close()
|
|
if radio_channel is not None: radio_channel.close()
|
|
os.close(journal)
|
|
|
|
|
|
def clock_refresh(owner,prepared,qualification,application,authority,app_peer,authority_peer,address,request):
|
|
"""Normal clock-only coordinator; retains every app/runtime/network lifetime.
|
|
|
|
The normal serialized caller has quiesced under its pending transaction lock.
|
|
Each phase is nonce/sequence/credential bound. Uncertainty is not redispatched.
|
|
"""
|
|
require(prepared.epoch['delegation'].get('allow_refresh') is True and owner.power_state is None,
|
|
'clock-only refresh not appointed or power transition active')
|
|
require(type(request) is dict and set(request)=={'status','sequence','nonce'} and request['status']=='CLOCK_REFRESH' and
|
|
type(request['sequence']) is int and request['sequence']==owner.clock_sequence+1 and request['sequence']<2**64 and
|
|
type(request['nonce']) is list and len(request['nonce'])==32 and any(request['nonce']) and
|
|
all(type(n) is int and 0<=n<=255 for n in request['nonce']), 'clock refresh replay/shape')
|
|
owner.clock_sequence=request['sequence']
|
|
token={'sequence':request['sequence'],'nonce':request['nonce']}
|
|
end=now()+SETUP_NS
|
|
def received(record,credentials,status,context=False):
|
|
value=json.loads(owner.receive(record,credentials['uid'],credentials['gid'],end),object_pairs_hook=exact_object)
|
|
require(type(value) is dict and set(value)==({'status','sequence','nonce','challenge'} if context else {'status','sequence','nonce'}) and
|
|
value['status']==status and value['sequence']==token['sequence'] and value['nonce']==token['nonce'],
|
|
'clock refresh phase/nonce mismatch')
|
|
return value
|
|
packet(authority.channel,{'command':'CLOCK_REFRESH',**token})
|
|
received(authority,authority_peer,'CLOCK_RETIRED')
|
|
packet(application.channel,{'status':'CLOCK_CONNECT',**token})
|
|
received(application,app_peer,'CLOCK_CONNECTED')
|
|
packet(authority.channel,{'command':'CLOCK_ACCEPT',**token})
|
|
context=received(authority,authority_peer,'CLOCK_CONTEXT',True)['challenge']
|
|
while 'collector' in owner.children.children and owner.tick() and now()<end:
|
|
time.sleep(.01)
|
|
require(owner.tick() and now()<end,'collector retirement unavailable')
|
|
owner.children.rearm_collector()
|
|
collector=start_collector(owner,qualification,prepared.value['binding'],bytes.fromhex(prepared.value['clock_domain']),
|
|
authority_peer,address,end,prepared.epoch,context)
|
|
owner.relay_collector(collector,authority,authority_peer['uid'],authority_peer['gid'],end)
|
|
received(authority,authority_peer,'CLOCK_AUTHORIZED')
|
|
owner.wait(collector,b'ADMIN_SENT',0,0,end)
|
|
owner.completed_one_shots.add('collector')
|
|
packet(collector.channel,b'RELEASE')
|
|
packet(application.channel,{'status':'CLOCK_AUTHORIZED',**token})
|
|
received(application,app_peer,'CLOCK_RESUMED')
|
|
print('C01_CLOCK_REFRESH_COMMITTED application_pid='+str(application.pid)+' epoch='+str(context.get('epoch')),flush=True)
|
|
|
|
|
|
def main():
|
|
require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'supervisor requires root')
|
|
require(LIBC.prctl(4, 0, 0, 0, 0) == 0 and LIBC.prctl(38, 1, 0, 0, 0) == 0,
|
|
'supervisor kernel protection unavailable')
|
|
close_except(set())
|
|
lock = LaunchLock()
|
|
owner = None
|
|
prepared = None
|
|
try:
|
|
owner = Supervisor(lock)
|
|
address = os.environ.get('NOTIFY_SOCKET')
|
|
if address:
|
|
require(address.startswith(('/', '@')), 'invalid service notification address')
|
|
owner.notifier = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
|
|
owner.notifier.connect('\0' + address[1:] if address.startswith('@') else address)
|
|
owner.notifier.setblocking(False)
|
|
value, app, authority, qualification = load_configuration()
|
|
prepared = Prepared(value)
|
|
acquire_admission_leases(owner, value, app, authority)
|
|
owner.power_lease = c01_power.Lease(value)
|
|
print('C01_LIFETIME_CAPABILITIES_ACQUIRED', flush=True)
|
|
maximum = prepared.epoch['delegation']['max_awake']
|
|
qualification.validate_epoch(maximum['secs'] * 1_000_000_000 + maximum['nanos'])
|
|
admit(value, app, authority, owner)
|
|
launch(owner, prepared, app, authority, qualification)
|
|
finally:
|
|
# Retain descriptors, payload and lock through ALL retirement, including
|
|
# stalled workers. Never put close(lock) in a deadline/error fallback.
|
|
if owner is not None:
|
|
owner.retire()
|
|
else:
|
|
lock.close()
|
|
if prepared is not None:
|
|
prepared.close()
|