1023 lines
54 KiB
Python
1023 lines
54 KiB
Python
#!/usr/bin/python3
|
|
"""C01 root access/retirement owner; fixed source, fixed policy, finite protocol.
|
|
|
|
`transition` is an OFFLINE guarded operation, never exposed on the socket.
|
|
`serve` issues revocable capabilities only with current kernel/policy evidence.
|
|
An interrupted transition or owner lifetime leaves a persistent restart fence.
|
|
Neither CLI installs itself, starts the application, nor bypasses admit().
|
|
"""
|
|
import array
|
|
import errno
|
|
import fcntl
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import select
|
|
import signal
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import types
|
|
|
|
# The -I entry never searches a writable directory or loads adjacent bytecode.
|
|
if __name__ == '__main__':
|
|
for name in ('c01_network', 'c01_access_policy', 'c01_migration_layout', 'c01_access_generation', 'c01_access_kernel', 'c01_children'):
|
|
path = Path(__file__).absolute().parent / (name + '.py')
|
|
for item in [*reversed(path.parents), path]:
|
|
meta = item.lstat()
|
|
if (meta.st_uid != 0 or meta.st_mode & 0o022 or stat.S_ISLNK(meta.st_mode) or
|
|
not (stat.S_ISREG(meta.st_mode) if item == path else stat.S_ISDIR(meta.st_mode))):
|
|
raise ValueError('access module lookup is not root protected')
|
|
module = types.ModuleType(name)
|
|
module.__file__ = str(path)
|
|
sys.modules[name] = module
|
|
exec(compile(path.read_bytes(), str(path), 'exec'), module.__dict__)
|
|
|
|
import c01_access_policy as policy_module
|
|
import c01_access_kernel as kernel
|
|
import c01_children as children_module
|
|
import c01_access_generation
|
|
|
|
require = policy_module.require
|
|
BASE = Path('/run/archipelago-c01')
|
|
SOCKET = BASE / 'runtime-access.sock'
|
|
MAX_PACKET = 8192
|
|
CHECK_NS = 5_000_000_000
|
|
WATCH_PACKET = struct.Struct('!8sQQQ')
|
|
OPERATIONS = ['freeze-capture', 'mask-stop-activation', 'replace-policy', 'kill-observe-empty',
|
|
'reload-ssh-restart-polkit', 'quarantine-specific-copies', 'verify-effective-policy-and-preservation']
|
|
ENV = {'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'LC_ALL': 'C', 'HOME': '/root'}
|
|
|
|
|
|
def send(channel, value):
|
|
raw = json.dumps(value, separators=(',', ':')).encode()
|
|
require(len(raw) <= MAX_PACKET and
|
|
channel.send(raw, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(raw), 'access send failed')
|
|
|
|
|
|
def receive(channel, credentials=None):
|
|
raw, ancillary, flags, _ = channel.recvmsg(MAX_PACKET + 1, socket.CMSG_SPACE(12) + socket.CMSG_SPACE(64),
|
|
socket.MSG_DONTWAIT | socket.MSG_CMSG_CLOEXEC)
|
|
peers, unwanted = [], False
|
|
for level, kind, data in ancillary:
|
|
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)
|
|
unwanted = True
|
|
elif (level, kind) == (socket.SOL_SOCKET, socket.SCM_CREDENTIALS) and len(data) == 12:
|
|
peers.append(struct.unpack('3i', data))
|
|
else:
|
|
unwanted = True
|
|
require(raw, 'access channel closed before protocol packet')
|
|
require(0 < len(raw) <= MAX_PACKET and not flags & ~socket.MSG_CMSG_CLOEXEC and not unwanted and
|
|
peers == ([] if credentials is None else [credentials]), 'access framing/credentials refused')
|
|
return json.loads(raw, object_pairs_hook=policy_module.exact)
|
|
|
|
|
|
def peer(channel):
|
|
return struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
|
|
|
|
def revoke(channel):
|
|
# shutdown revokes aliases too, including any accidental in-process duplicate.
|
|
try:
|
|
channel.shutdown(socket.SHUT_RDWR)
|
|
except OSError:
|
|
pass
|
|
channel.close()
|
|
|
|
|
|
def notify_ready():
|
|
address = os.environ.get('NOTIFY_SOCKET')
|
|
if address:
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as channel:
|
|
channel.connect('\0' + address[1:] if address.startswith('@') else address)
|
|
channel.send(b'READY=1\nSTATUS=Live access policy checked; guarded lease listener available')
|
|
|
|
|
|
def fixed_command(argv):
|
|
"""Only called by the bounded, cgroup-owned root checker/transition child.
|
|
|
|
No caller chooses argv. A stuck spawn/read/reap strands that helper, not the
|
|
MainPID. MainPID revokes and keeps its lock and Children registry until reaped.
|
|
"""
|
|
import resource
|
|
import tempfile
|
|
# Regular bounded output avoids pipe-EOF descendants and unbounded buffering.
|
|
with tempfile.TemporaryFile() as output:
|
|
result = subprocess.run(argv, stdin=subprocess.DEVNULL, stdout=output, stderr=subprocess.DEVNULL,
|
|
env=ENV, timeout=5, check=False,
|
|
preexec_fn=lambda: resource.setrlimit(resource.RLIMIT_FSIZE, (1024 * 1024, 1024 * 1024)))
|
|
output.seek(0)
|
|
raw = output.read(1024 * 1024 + 1)
|
|
require(len(raw) <= 1024 * 1024, 'policy probe output bound')
|
|
return result.returncode, raw.decode()
|
|
|
|
|
|
def semantic_checks(policy):
|
|
import pwd
|
|
v = policy.value
|
|
shadow = {row.split(':')[0]: row.split(':')[1]
|
|
for row in policy_module.read(Path('/etc/shadow'), True).decode().splitlines()}
|
|
operator = pwd.getpwnam(v['operator']['name'])
|
|
require((operator.pw_uid, operator.pw_gid) == (1001, 1001) and
|
|
operator.pw_shell not in ('/sbin/nologin', '/usr/sbin/nologin', '/bin/false') and
|
|
((bool(shadow[operator.pw_name]) and not shadow[operator.pw_name].startswith(('!', '*'))) or
|
|
v.get('operator_sudo_authentication') == 'preserve-verified-noninteractive'),
|
|
'separate operator access not preserved')
|
|
require(fixed_command(['/usr/sbin/visudo', '-c'])[0] == 0, 'installed sudo policy is invalid')
|
|
nss = policy_module.read(Path('/etc/nsswitch.conf')).decode().splitlines()
|
|
sudo_sources = [row.split(':', 1)[1].split('#', 1)[0].split() for row in nss
|
|
if row.split(':', 1)[0].strip() == 'sudoers']
|
|
require(not sudo_sources or sudo_sources == [['files']], 'external sudo policy source refused')
|
|
# Exact sudoers bytes plus syntax success distinguish an intended denial
|
|
# from a missing/broken sudo installation. NSS/plugin inventory is pinned.
|
|
for account in v['accounts'].values():
|
|
entry = pwd.getpwnam(account['name'])
|
|
require((entry.pw_uid, entry.pw_gid) == (account['uid'], account['gid']) and
|
|
entry.pw_shell in ('/sbin/nologin', '/usr/sbin/nologin', '/bin/false') and
|
|
os.getgrouplist(entry.pw_name, entry.pw_gid) == [entry.pw_gid] and
|
|
sum(p.pw_uid == entry.pw_uid for p in pwd.getpwall()) == 1 and
|
|
shadow[entry.pw_name].startswith('!'), 'managed login/credential policy changed')
|
|
code, listing = fixed_command(['/usr/bin/sudo', '-n', '-l', '-U', entry.pw_name])
|
|
# Some sudo versions return zero for a successfully produced *empty*
|
|
# listing. Require the exact C-locale no-permission result, not status
|
|
# alone (and never interpret an authentication/plugin failure as denial).
|
|
denial = f'User {entry.pw_name} is not allowed to run sudo on {os.uname().nodename}.'
|
|
require(code in (0, 1) and listing.strip() == denial, 'managed principal sudo denial unproved')
|
|
ssh_checks(policy)
|
|
# A rule returning YES before the C01 rule would bypass it. Refuse every
|
|
# lexically earlier/equal competing rule in both polkit search roots.
|
|
rule = '00-archipelago-c01.rules'
|
|
for directory in ('/etc/polkit-1/rules.d', '/run/polkit-1/rules.d',
|
|
'/usr/local/share/polkit-1/rules.d', '/usr/share/polkit-1/rules.d'):
|
|
if Path(directory).exists():
|
|
policy_module.protected(Path(directory), directory=True)
|
|
for path in Path(directory).glob('*.rules'):
|
|
policy_module.tree_digest(path)
|
|
require(path.name > rule or path == policy_module.TARGETS['polkit'], 'earlier polkit rule can grant access')
|
|
for unit in v['restart_units']:
|
|
path = Path('/etc/systemd/system') / unit
|
|
policy_module.protected(path.parent, directory=True)
|
|
require(path.is_symlink() and os.readlink(path) == '/dev/null', 'legacy activation path is not masked')
|
|
# An allow file takes precedence over a deny file. Jobs already in spools
|
|
# must be explicitly retired, not merely prevented from new submission.
|
|
for path in (Path('/etc/cron.allow'), Path('/etc/at.allow')):
|
|
if path.exists():
|
|
require(not set(policy.names) & set(policy_module.read(path).decode().splitlines()), 'job allowlist grants access')
|
|
for name in policy.names:
|
|
require(not any((Path(root) / name).exists() for root in ('/var/spool/cron', '/var/spool/cron/crontabs')),
|
|
'managed crontab remains active')
|
|
for directory in (Path('/var/spool/at'), Path('/var/spool/cron/atjobs')):
|
|
if directory.exists():
|
|
managed = {item['uid'] for item in v['accounts'].values()}
|
|
require(not any(path.lstat().st_uid in managed for path in directory.iterdir()),
|
|
'managed queued at job remains active')
|
|
# Subordinate IDs and original paths are preserved without chown/copy.
|
|
for file, key in (('/etc/subuid', 'subuids'), ('/etc/subgid', 'subgids')):
|
|
rows = [row.split(':') for row in policy_module.read(Path(file)).decode().splitlines()]
|
|
ranges = [[int(row[1]), int(row[2])] for row in rows if row[0] == policy.runtime['runtime_user']]
|
|
require(ranges == policy.runtime[key], 'original subordinate allocation changed')
|
|
|
|
|
|
def ssh_effective(text):
|
|
# OpenSSH emits list-valued settings (including DenyUsers/AllowUsers) as
|
|
# repeated rows. Never discard earlier entries with dict(row.split(...)).
|
|
result = {}
|
|
for row in text.splitlines():
|
|
if ' ' in row:
|
|
name, value = row.split(' ', 1)
|
|
result.setdefault(name, []).append(value)
|
|
return result
|
|
|
|
|
|
def ssh_checks(policy, staged=False):
|
|
"""Global-only finite policy plus actual sshd parsing, including remote IPs.
|
|
|
|
The byte-preserved baseline is the coordinator's verified administrator
|
|
route. This checks configuration preservation, not an SSH authentication UAT.
|
|
"""
|
|
target = policy_module.STAGED / 'sshd' if staged else policy_module.TARGETS['sshd']
|
|
post = policy_module.read(target)
|
|
policy_module.ssh_baseline(post, policy.names, policy.value['files']['sshd']['before'])
|
|
require(fixed_command(['/usr/sbin/sshd', '-t', '-f', str(target)])[0] == 0, 'staged/current SSH syntax refused')
|
|
for address in ('127.0.0.1', '198.51.100.23', '2001:db8::23'):
|
|
for name in [*policy.names, policy.value['operator']['name']]:
|
|
context = f'user={name},host=admin.example,addr={address}'
|
|
code, text = fixed_command(['/usr/sbin/sshd', '-T', '-f', str(target), '-C', context])
|
|
effective = ssh_effective(text)
|
|
denied = [name for row in effective.get('denyusers', []) for name in row.split()]
|
|
require(code == 0 and denied == policy.names,
|
|
'global effective SSH denial differs')
|
|
if staged and name == policy.value['operator']['name']:
|
|
code, old_text = fixed_command(['/usr/sbin/sshd', '-T', '-f', str(policy_module.TARGETS['sshd']),
|
|
'-C', context])
|
|
old = ssh_effective(old_text)
|
|
require(code == 0 and {k: v for k, v in effective.items() if k != 'denyusers'} ==
|
|
{k: v for k, v in old.items() if k != 'denyusers'}, 'administrator SSH route changed')
|
|
|
|
|
|
def completed_history(value, policy):
|
|
require(value['state'] in ('retired', 'owner-active') and value['configuration_sha256'] == policy.digest,
|
|
'no completed operation history to requalify')
|
|
evidence = value['evidence']
|
|
require(type(evidence) is dict and evidence.get('operations') == OPERATIONS and
|
|
type(evidence.get('writers')) is list and type(evidence.get('copies')) is list,
|
|
'incomplete operation provenance')
|
|
expected = {(int(pid), start, spec['path']) for spec in policy.value['cohorts']
|
|
for pid, start in spec['members'].items()}
|
|
writers = evidence['writers']
|
|
require(len(writers) == len(expected) and all(any(
|
|
item['pid'] == pid and item['start'] == start and Path(item['group']).is_relative_to(Path(group))
|
|
for item in writers) for pid, start, group in expected), 'retirement writer provenance differs')
|
|
copies = [{'index': index, 'operation': 'quarantined-known-copy' if item['disposition'] ==
|
|
'quarantine-known-copy' else 'verified-exclusive-backing', 'identity': item['identity']}
|
|
for index, item in enumerate(policy.value['copies'])]
|
|
require(evidence['copies'] == copies, 'retirement copy provenance differs')
|
|
if policy.value['version']==3:
|
|
plan=policy.value['migration_layout']
|
|
expected_layout={'source_identity':plan['source_identity'],'target_identity':plan['target_identity'],
|
|
'moved':[{k:item[k] for k in ('name','identity','mode_after','tree_metadata_sha256')}
|
|
for item in plan['entries'] if item['disposition']=='private']}
|
|
require(evidence.get('layout')==expected_layout,'missing exact layout preservation evidence')
|
|
|
|
|
|
class Journal:
|
|
"""One persistent inode/lock and a conservative crash fence, never a grant."""
|
|
def __init__(self, policy):
|
|
self.policy = policy
|
|
policy_module.protected(policy_module.STATE, private=True, directory=True)
|
|
self.path = policy_module.STATE / 'journal.json'
|
|
self.lock = policy_module.STATE / 'owner.lock'
|
|
self.fd = os.open(self.lock, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
|
|
try:
|
|
meta = os.fstat(self.fd)
|
|
require(meta.st_uid == 0 and stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1 and
|
|
stat.S_ISREG(meta.st_mode), 'unprotected access lock')
|
|
fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
require(self.lock.lstat().st_ino == meta.st_ino, 'access lock replaced')
|
|
except BaseException:
|
|
os.close(self.fd)
|
|
raise
|
|
self.boot = Path('/proc/sys/kernel/random/boot_id').read_text().strip()
|
|
|
|
def record(self, state, evidence=None):
|
|
owner = kernel.process(os.getpid())
|
|
policy_module.atomic(self.path, json.dumps({'version': 1, 'state': state,
|
|
'configuration_sha256': self.policy.digest, 'boot': self.boot,
|
|
'owner': {'pid': os.getpid(), 'start': owner['start'], 'group': owner['group']},
|
|
'evidence': evidence}, separators=(',', ':')).encode())
|
|
|
|
def begin_transition(self):
|
|
require(not self.path.exists(), 'prior access transition/lifetime requires offline reconciliation')
|
|
self.record('transition-pending')
|
|
|
|
def begin_serving(self):
|
|
value, _ = policy_module.json_read(self.path)
|
|
completed_history(value, self.policy)
|
|
require(value['state'] == 'retired' and value['configuration_sha256'] == self.policy.digest and
|
|
value['boot'] == self.boot, 'no reconciled same-boot retirement; cached state cannot admit')
|
|
self.record('owner-active', value['evidence'])
|
|
|
|
def pin_endpoint(self):
|
|
value, _ = policy_module.json_read(self.path)
|
|
meta = SOCKET.lstat()
|
|
self.record('owner-active', {**value['evidence'], 'endpoint': policy_module.identity(meta)})
|
|
|
|
def close(self):
|
|
os.close(self.fd)
|
|
|
|
|
|
def preflight(policy):
|
|
policy.preserved()
|
|
policy.preflight_files()
|
|
if policy.value['version']==3:
|
|
import c01_migration_layout
|
|
c01_migration_layout.preflight(policy)
|
|
for unit in policy.value['restart_units']:
|
|
path = Path('/etc/systemd/system') / unit
|
|
policy_module.protected(path.parent, directory=True)
|
|
require(not os.path.lexists(path), 'unit mask preflight conflict: ' + unit)
|
|
ssh_checks(policy, staged=True)
|
|
require(fixed_command(['/usr/sbin/visudo', '-c', '-f', str(policy_module.STAGED / 'sudoers')])[0] == 0,
|
|
'staged sudo syntax refused')
|
|
cohorts = []
|
|
try:
|
|
paths = [policy_module.canonical(spec['path']) for spec in policy.value['cohorts']]
|
|
require(all(not a.is_relative_to(b) and not b.is_relative_to(a)
|
|
for index, a in enumerate(paths) for b in paths[index + 1:]), 'overlapping cohort inventories')
|
|
for spec in policy.value['cohorts']:
|
|
cohorts.append(kernel.Cohort(spec, policy.runtime))
|
|
managed = {account['uid'] for account in policy.value['accounts'].values()}
|
|
classified = {int(pid) for spec in policy.value['cohorts'] for pid in spec['members']}
|
|
require(all(not managed.intersection(item['uids']) or item['pid'] in classified
|
|
for item in kernel.all_processes()), 'unclassified preflight credential-bearing process')
|
|
finally:
|
|
for cohort in cohorts:
|
|
cohort.close()
|
|
|
|
|
|
def retirement(policy, progress, checkpoint=None):
|
|
"""Guarded finite transition. Caller retains the helper's full cgroup."""
|
|
cohorts = []
|
|
try:
|
|
preflight(policy)
|
|
for spec in policy.value['cohorts']:
|
|
cohorts.append(kernel.Cohort(spec, policy.runtime))
|
|
for cohort in cohorts:
|
|
cohort.freeze()
|
|
deadline = kernel.now() + CHECK_NS
|
|
while not all(c.events().get('frozen') == '1' for c in cohorts):
|
|
require(kernel.now() < deadline, 'legacy cohort freeze incomplete')
|
|
time.sleep(.02)
|
|
writers = [item for cohort in cohorts for item in cohort.capture()]
|
|
classified = {item['pid'] for item in writers}
|
|
managed = {item['uid'] for item in policy.value['accounts'].values()}
|
|
require(all(not managed.intersection(item['uids']) or item['pid'] in classified
|
|
for item in kernel.all_processes()), 'unclassified old credential-bearing process')
|
|
progress('frozen-inventoried-cohorts')
|
|
# Masks are persistent and precede policy replacement and old-key custody.
|
|
for unit in policy.value['restart_units']:
|
|
path = Path('/etc/systemd/system') / unit
|
|
policy_module.protected(path.parent, directory=True)
|
|
require(not os.path.lexists(path), 'unit mask requires reviewed removal of existing override')
|
|
os.symlink('/dev/null', path)
|
|
require(fixed_command(['/usr/bin/systemctl', 'daemon-reload'])[0] == 0, 'mask reload failed')
|
|
# Stop requests cancel queued activations; do not wait for process exit.
|
|
require(fixed_command(['/usr/bin/systemctl', '--no-block', 'stop', *policy.value['restart_units']])[0] == 0,
|
|
'legacy activation stop failed')
|
|
progress('masked-and-stopped-activation')
|
|
policy.install_files()
|
|
progress('installed-exact-access-policy')
|
|
# File parsing alone does not retire a daemon's old policy/cache. These
|
|
# fixed service operations are part of the guarded transaction. Existing
|
|
# operator SSH sessions are preserved by reload (never ssh restart).
|
|
require(fixed_command(['/usr/bin/systemctl', 'reload', 'ssh.service'])[0] == 0,
|
|
'SSH policy reload failed')
|
|
require(fixed_command(['/usr/bin/systemctl', 'restart', 'polkit.service'])[0] == 0,
|
|
'polkit policy/cache retirement failed')
|
|
progress('reloaded-ssh-retired-polkit-cache')
|
|
for cohort in cohorts:
|
|
cohort.kill()
|
|
while not all(c.retired() for c in cohorts):
|
|
# No false completion deadline: MainPID owns this helper and its
|
|
# timeout fences admission; unresolved legacy cohorts stay journaled.
|
|
time.sleep(.02)
|
|
progress('kernel-observed-old-writer-exit')
|
|
layout=None
|
|
if policy.value['version']==3:
|
|
require(checkpoint is not None,'layout migration needs a durable retired-writer checkpoint')
|
|
checkpoint('writers-retired',{'writers':writers})
|
|
import c01_migration_layout
|
|
layout=c01_migration_layout.apply_retired(policy)
|
|
copies = kernel.quarantine(policy)
|
|
policy.verify_files()
|
|
policy.verify_trees()
|
|
policy.preserved()
|
|
semantic_checks(policy)
|
|
kernel.verify_private(policy)
|
|
require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),
|
|
'legacy credential cohort respawned during transition')
|
|
evidence={'writers': writers, 'copies': copies, 'operations': OPERATIONS}
|
|
if layout is not None:evidence['layout']=layout
|
|
return evidence
|
|
finally:
|
|
# Never thaw surviving old writers or restore old privileges on failure.
|
|
for cohort in cohorts:
|
|
cohort.close()
|
|
|
|
|
|
def check_current(policy, supervisor):
|
|
for label, operation in (('files', policy.verify_files), ('trees', policy.verify_trees),
|
|
('preserved', policy.preserved), ('semantics', lambda: semantic_checks(policy)),
|
|
('private', lambda: kernel.verify_private(policy))):
|
|
started=kernel.now()
|
|
operation()
|
|
elapsed=kernel.now()-started
|
|
if elapsed > 500_000_000:
|
|
print('C01_ACCESS_SLOW_CHECK: '+label+' '+str(elapsed),flush=True)
|
|
# A reused same inode could still contain an unresolved original cohort.
|
|
# Even an empty old group must be retired by its systemd owner before serve.
|
|
for spec in policy.value['cohorts']:
|
|
kernel.replacement_tree(spec, policy.runtime)
|
|
retired = False
|
|
active = supervisor
|
|
if supervisor is not None and supervisor.get('retiring'):
|
|
active = None
|
|
group = Path(policy.value['supervisor_cgroup'])
|
|
# Keep the reserved lifetime until BOTH direct owner exit and complete
|
|
# kernel role retirement. No adoption of an empty-but-unremoved role.
|
|
retired = (not group.exists() or (not kernel.members(group) and
|
|
not any((group / ('c01-' + role)).exists() for role in children_module.ROLES)))
|
|
elif supervisor is not None:
|
|
try:
|
|
current = kernel.process(supervisor['pid'])
|
|
except (FileNotFoundError, ProcessLookupError):
|
|
# MainPID's retained pidfd still owns this identity. A clean owner
|
|
# exit may race this check; no remaining managed process is allowed.
|
|
active = None
|
|
else:
|
|
require(current['start'] == supervisor['start'] and current['uids'] == [0] * 4 and
|
|
current['gids'] == [0] * 4 and
|
|
Path(current['group']).is_relative_to(Path(policy.value['supervisor_cgroup'])),
|
|
'exclusive supervisor changed')
|
|
kernel.verify_supervisor(current, policy)
|
|
# A reserved candidate has no authority to launch roles until publication.
|
|
# Thus an existing orphan cannot acquire legitimacy by asking for a grant.
|
|
evidence = kernel.verify_processes(policy, None if active and active.get('pending') else active)
|
|
return {'runtime_processes': len(evidence), 'supervisor_retired': retired}
|
|
|
|
|
|
def checker(control, policy):
|
|
# This process is forked BEFORE public listeners/leases exist. Inheritance
|
|
# cannot keep a public provider endpoint alive after MainPID death.
|
|
while True:
|
|
ready, _, _ = select.select([control], [], [], 1)
|
|
if not ready:
|
|
continue
|
|
request = receive(control)
|
|
require(set(request) == {'challenge', 'supervisor', 'generation', 'access_generation_head'}, 'private checker protocol')
|
|
observed = kernel.now()
|
|
try:
|
|
policy.generations.refresh(request['access_generation_head'])
|
|
evidence = check_current(policy, request['supervisor'])
|
|
except (OSError, ValueError, KeyError) as error:
|
|
# Fixed policy/kernel diagnostics only; no file content or credentials.
|
|
send(control, {'challenge': request['challenge'], 'generation': request['generation'],
|
|
'failed': type(error).__name__, 'reason': str(error)[:256]})
|
|
continue # An obsolete generation cannot poison the next lifecycle.
|
|
send(control, {'challenge': request['challenge'], 'generation': request['generation'],
|
|
'observed': observed, **evidence})
|
|
|
|
|
|
def watch_packet(kind, generation, deadline, token=0):
|
|
return WATCH_PACKET.pack(kind, generation, deadline, token)
|
|
|
|
|
|
def watch_evidence(raw, generation, evidence_deadline, stamp):
|
|
require(len(raw) == WATCH_PACKET.size, 'access observer packet size')
|
|
kind, current, expiry, token = WATCH_PACKET.unpack(raw)
|
|
require(kind in (b'PULSE\0\0\0', b'HOLD\0\0\0\0') and current >= generation and
|
|
evidence_deadline <= expiry and stamp < expiry <= stamp + CHECK_NS,
|
|
'access observer evidence expired/regressed')
|
|
return kind.rstrip(b'\0'), current, expiry, token
|
|
|
|
|
|
def watchdog(control, parent_pidfd):
|
|
"""Independent BOOTTIME observer; no public socket exists at its fork.
|
|
|
|
The frozen runtime-access-v1 protocol has no renewal messages. MainPID sends
|
|
an explicit revocation alias BEFORE each grant. This observer can shutdown
|
|
those aliases even if MainPID is stuck in an uninterruptible kernel syscall;
|
|
it does not rely on SIGKILL completion. PDEATHSIG kills this no-I/O/no-fork
|
|
observer on MainPID death, so aliases cannot conceal a dead provider.
|
|
"""
|
|
deadline = kernel.now() + CHECK_NS
|
|
evidence_deadline = 0
|
|
generation = 0
|
|
held = set()
|
|
try:
|
|
while kernel.now() < deadline:
|
|
ready, _, _ = select.select([control, *held], [], [], .05)
|
|
if kernel.now() >= deadline:
|
|
break # A queued renewal cannot resurrect already expired evidence.
|
|
if control in ready:
|
|
raw, ancillary, flags, _ = control.recvmsg(WATCH_PACKET.size, socket.CMSG_SPACE(16 * 4), socket.MSG_CMSG_CLOEXEC)
|
|
rights, unwanted = [], False
|
|
for level, kind, data in ancillary:
|
|
if (level, kind) == (socket.SOL_SOCKET, socket.SCM_RIGHTS):
|
|
values = array.array('i')
|
|
values.frombytes(data[:len(data) - len(data) % values.itemsize])
|
|
rights.extend(values)
|
|
else:
|
|
unwanted = True
|
|
try:
|
|
kind, generation, evidence_deadline, token = watch_evidence(raw, generation, evidence_deadline, kernel.now())
|
|
require(not unwanted and not flags & ~socket.MSG_CMSG_CLOEXEC and
|
|
((kind == b'PULSE' and not rights and token == 0) or
|
|
(kind == b'HOLD' and len(rights) == 1 and token > 0)),
|
|
'access observer control lost')
|
|
if rights:
|
|
channel = socket.socket(fileno=rights.pop())
|
|
held.add(channel)
|
|
require(len(held) <= 32, 'access revocation alias bound')
|
|
deadline = min(evidence_deadline, kernel.now() + CHECK_NS)
|
|
if kind == b'PULSE':
|
|
send(control, {'alive': kernel.now()})
|
|
else:
|
|
send(control, {'held': token})
|
|
finally:
|
|
for fd in rights:
|
|
revoke(socket.socket(fileno=fd))
|
|
for channel in list(held):
|
|
if channel in ready:
|
|
revoke(channel)
|
|
held.remove(channel)
|
|
for channel in held:
|
|
revoke(channel)
|
|
held.clear()
|
|
finally:
|
|
for channel in held:
|
|
revoke(channel)
|
|
signal.pidfd_send_signal(parent_pidfd, signal.SIGKILL)
|
|
|
|
|
|
def start_watchdog(children):
|
|
fd = os.pidfd_open(os.getpid())
|
|
try:
|
|
return children.spawn('authority', lambda control: watchdog(control, fd), keep=(fd,))
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
class Leases:
|
|
"""Authenticated challenge/lifetime state; no persisted admission Boolean."""
|
|
def __init__(self, policy, guard):
|
|
self.policy = policy
|
|
self.guard = guard
|
|
self.clients = {}
|
|
self.nonces = set()
|
|
self.supervisor = None
|
|
self.retiring = None
|
|
self.last_observed = 0
|
|
self.generation = 0
|
|
self.observed_generation = 0
|
|
self.next_grant = 0
|
|
self.failed = False
|
|
# No fork is needed once serving. Defense in depth for a future helper:
|
|
# close aliases in the child; shutdown would incorrectly revoke parent.
|
|
os.register_at_fork(after_in_child=self.close_in_child)
|
|
|
|
def close_in_child(self):
|
|
for channel, entry in self.clients.items():
|
|
channel.close()
|
|
os.close(entry['pidfd'])
|
|
if self.retiring is not None:
|
|
os.close(self.retiring['pidfd'])
|
|
|
|
def fresh(self):
|
|
return not self.failed and 0 <= kernel.now() - self.last_observed < CHECK_NS
|
|
|
|
def accept(self, channel, request):
|
|
pid, uid, gid = peer(channel)
|
|
require((uid, gid) == (0, 0) and pid > 1 and self.fresh(), 'access peer/current evidence refused')
|
|
require(type(request) is dict and set(request) == {'protocol', 'configuration_sha256', 'nonce'} and
|
|
policy_module.hex_digest(request['nonce']) and request['nonce'] not in self.nonces and
|
|
len(self.nonces) < 4096 and len(self.clients) < 16, 'access challenge bound/replay')
|
|
protocol = request['protocol']
|
|
require(protocol in ('runtime-access-v1', 'supervisor-access-v1'), 'unknown access protocol')
|
|
role = 'broker' if protocol == 'runtime-access-v1' else 'supervisor'
|
|
if role == 'supervisor':
|
|
require(self.policy.generations.value['pending'] is None, 'launch generation transition pending')
|
|
digest = self.policy.value['runtime_sha256'] if role == 'broker' else self.policy.generations.value['active_launch_sha256']
|
|
require(request['configuration_sha256'] == digest, 'access configuration digest mismatch')
|
|
pidfd = os.pidfd_open(pid)
|
|
try:
|
|
item = kernel.process(pid)
|
|
require(item['uids'] == [0] * 4 and item['gids'] == [0] * 4 and
|
|
Path(item['group']).is_relative_to(Path(self.policy.value[role + '_cgroup'])) and
|
|
not select.select([pidfd], [], [], 0)[0], 'root service identity mismatch')
|
|
require(not any(entry['role'] == role for entry in self.clients.values()), 'exclusive access role already held')
|
|
require(role != 'supervisor' or self.retiring is None, 'prior supervisor retirement remains owned')
|
|
self.nonces.add(request['nonce'])
|
|
self.clients[channel] = {'pidfd': pidfd, 'role': role, 'pid': pid, 'start': item['start']}
|
|
if role == 'supervisor':
|
|
self.generation += 1
|
|
self.supervisor = {'pid': pid, 'start': item['start'], 'pending': True}
|
|
self.clients[channel]['request'] = request
|
|
else:
|
|
self.publish(channel, request)
|
|
except BaseException:
|
|
if channel not in self.clients:
|
|
os.close(pidfd)
|
|
raise
|
|
|
|
def publish(self, channel, request):
|
|
require(self.fresh(), 'expired access grant')
|
|
self.next_grant += 1
|
|
packet = watch_packet(b'HOLD', self.observed_generation, self.last_observed + CHECK_NS, self.next_grant)
|
|
require(self.guard.sendmsg([packet], [(socket.SOL_SOCKET, socket.SCM_RIGHTS,
|
|
array.array('i', [channel.fileno()]))], socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(packet),
|
|
'independent access revocation capability delivery failed')
|
|
self.clients[channel]['grant'] = (self.next_grant, request)
|
|
|
|
def confirm(self, token):
|
|
require(type(token) is int, 'invalid watchdog handoff acknowledgment')
|
|
for channel, entry in self.clients.items():
|
|
if entry.get('grant', (None,))[0] == token:
|
|
require(self.fresh() and not select.select([channel, entry['pidfd']], [], [], 0)[0],
|
|
'expired/departed candidate before acknowledged grant')
|
|
_, request = entry.pop('grant')
|
|
send(channel, request)
|
|
return
|
|
# A disconnected candidate may have retired between HOLD and its ack.
|
|
require(0 < token <= self.next_grant, 'unknown watchdog handoff acknowledgment')
|
|
|
|
def checked(self, response, checked_supervisor):
|
|
if response['generation'] != self.generation:
|
|
return False
|
|
if 'failed' in response:
|
|
raise ValueError('access checker refused: ' + response['failed'] + ': ' + response['reason'])
|
|
require(not self.last_observed or self.fresh(), 'prior evidence expired before checker delivery')
|
|
self.last_observed = response['observed']
|
|
self.observed_generation = response['generation']
|
|
self.settle_supervisor(response['supervisor_retired'], checked_supervisor)
|
|
for channel, entry in self.clients.items():
|
|
if 'request' in entry:
|
|
require(checked_supervisor == self.supervisor and not select.select(
|
|
[channel, entry['pidfd']], [], [], 0)[0], 'candidate supervisor departed before grant')
|
|
self.publish(channel, entry.pop('request'))
|
|
self.supervisor = {'pid': entry['pid'], 'start': entry['start']}
|
|
# The candidate scan is consumed before handoff. Subsequent
|
|
# scans use active even if the watchdog ack is still pending;
|
|
# no older pending scan may fail after the grant is published.
|
|
self.generation += 1
|
|
return True
|
|
|
|
def poll(self):
|
|
require(self.fresh(), 'access evidence deadline expired')
|
|
for channel, entry in list(self.clients.items()):
|
|
if select.select([channel, entry['pidfd']], [], [], 0)[0]:
|
|
if entry['role'] != 'supervisor':
|
|
raise ValueError('runtime access consumer exited/disconnected/sent unsolicited data')
|
|
# Normal manager retirement must not take persistent runtime
|
|
# services with it. Any remaining managed role is still rejected
|
|
# by the checker; no replacement is issued over retained groups.
|
|
revoke(channel)
|
|
del self.clients[channel]
|
|
self.retiring = entry
|
|
self.generation += 1
|
|
self.supervisor = {'pid': entry['pid'], 'start': entry['start'], 'retiring': True}
|
|
|
|
def settle_supervisor(self, observed_retired, checked_supervisor):
|
|
if (self.retiring is not None and observed_retired and checked_supervisor == self.supervisor and
|
|
select.select([self.retiring['pidfd']], [], [], 0)[0]):
|
|
os.close(self.retiring['pidfd'])
|
|
self.retiring = None
|
|
self.supervisor = None
|
|
self.generation += 1
|
|
|
|
def stop(self):
|
|
self.failed = True
|
|
for channel, entry in self.clients.items():
|
|
revoke(channel)
|
|
os.close(entry['pidfd'])
|
|
self.clients.clear()
|
|
if self.retiring is not None:
|
|
os.close(self.retiring['pidfd'])
|
|
self.retiring = None
|
|
self.supervisor = None
|
|
|
|
|
|
def serve(policy, journal, children):
|
|
journal.begin_serving() # Crash fence before any checker or public capability.
|
|
record = children.spawn('host-io', lambda control: checker(control, policy))
|
|
guard = start_watchdog(children)
|
|
generation_worker = children.spawn('application', lambda control: c01_access_generation.worker(
|
|
control, policy.generations, policy.value['supervisor_cgroup'], send, receive))
|
|
generation_pending = None
|
|
record.channel.setblocking(False)
|
|
leases = Leases(policy, guard.channel)
|
|
listener = None
|
|
pending = {}
|
|
challenge = None
|
|
checked_supervisor = None
|
|
checked_generation = None
|
|
deadline = kernel.now() + CHECK_NS
|
|
startup_deadline = deadline
|
|
pulse = 0
|
|
guard_observed = kernel.now()
|
|
try:
|
|
while not children.stopping:
|
|
children.poll()
|
|
require('host-io' in children.children and 'authority' in children.children, 'access checker/watchdog exited')
|
|
stamp = kernel.now()
|
|
if stamp - pulse >= 250_000_000:
|
|
packet = watch_packet(b'PULSE', leases.observed_generation,
|
|
leases.last_observed + CHECK_NS if leases.last_observed else startup_deadline)
|
|
require(guard.channel.send(packet, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) == len(packet),
|
|
'access watchdog delivery failed')
|
|
pulse = stamp
|
|
if challenge is None and (generation_pending is None or not generation_pending[1]):
|
|
challenge = os.urandom(32).hex()
|
|
checked_supervisor = leases.supervisor
|
|
checked_generation = leases.generation
|
|
send(record.channel, {'challenge': challenge, 'supervisor': checked_supervisor,
|
|
'generation': checked_generation, 'access_generation_head':policy.generations.anchor})
|
|
deadline = stamp + CHECK_NS
|
|
require(stamp < deadline, 'access checker stalled')
|
|
require(stamp - guard_observed < CHECK_NS, 'independent access observer stalled')
|
|
ready, _, _ = select.select([record.channel, guard.channel, generation_worker.channel, *pending,
|
|
*([] if listener is None else [listener])], [], [], .05)
|
|
if generation_worker.channel in ready:
|
|
response=receive(generation_worker.channel,(generation_worker.pid,0,0))
|
|
require(generation_pending is not None,'unsolicited generation result')
|
|
channel,mutating=generation_pending
|
|
generation_pending=None
|
|
if set(response)=={'result','head','state'}:
|
|
require(set(response['state'])=={'active_launch_sha256','active_workspace_digest','pending'},'generation snapshot')
|
|
policy.generations.anchor=response['head']
|
|
policy.generations.value.update(response['state'])
|
|
if leases.supervisor is not None or leases.retiring is not None:
|
|
response['result']['quiescent']=False
|
|
try:
|
|
send(channel,response['result'])
|
|
except OSError:
|
|
pass # lost ack is inspected by exact record, never repeated
|
|
else:
|
|
require(response=={'failed':True},'generation helper outcome')
|
|
channel.close()
|
|
if mutating:
|
|
leases.generation+=1
|
|
if guard.channel in ready:
|
|
response = receive(guard.channel, (guard.pid, 0, 0))
|
|
if set(response) == {'held'}:
|
|
leases.confirm(response['held'])
|
|
else:
|
|
require(set(response) == {'alive'} and type(response['alive']) is int and
|
|
0 <= kernel.now() - response['alive'] < CHECK_NS, 'access observer heartbeat refused')
|
|
guard_observed = response['alive']
|
|
if record.channel in ready:
|
|
response = receive(record.channel, (record.pid, 0, 0))
|
|
require(set(response) in ({'challenge', 'generation', 'failed', 'reason'},
|
|
{'challenge', 'generation', 'observed', 'runtime_processes', 'supervisor_retired'}) and
|
|
response['challenge'] == challenge and response['generation'] == checked_generation,
|
|
'access checker generation refused')
|
|
if 'observed' in response:
|
|
require(type(response['observed']) is int and
|
|
0 <= kernel.now() - response['observed'] < CHECK_NS, 'access checker evidence refused')
|
|
accepted = leases.checked(response, checked_supervisor)
|
|
challenge = None
|
|
if listener is None and accepted:
|
|
policy_module.protected(BASE, directory=True)
|
|
listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_CLOEXEC)
|
|
listener.bind(str(SOCKET)) # Only offline requalification may retire a stale socket.
|
|
os.chmod(SOCKET, 0o600)
|
|
journal.pin_endpoint()
|
|
listener.listen(16)
|
|
listener.setblocking(False)
|
|
notify_ready()
|
|
if listener is not None:
|
|
leases.poll()
|
|
if listener in ready:
|
|
channel, _ = listener.accept()
|
|
channel.setblocking(False)
|
|
if len(pending) >= 16 or peer(channel)[1:] != (0, 0):
|
|
channel.close()
|
|
else:
|
|
pending[channel] = kernel.now() + 1_000_000_000
|
|
for channel, expiry in list(pending.items()):
|
|
if kernel.now() >= expiry:
|
|
channel.close()
|
|
del pending[channel]
|
|
elif channel in ready:
|
|
try:
|
|
request=receive(channel)
|
|
require(type(request) is dict, 'access request object required')
|
|
if request.get('protocol')=='d09-access-generation-v1':
|
|
pid,uid,gid=peer(channel)
|
|
current=kernel.process(pid)
|
|
code,main=fixed_command(['/usr/bin/systemctl','show','archipelago-d09-lifecycle.service','-p','MainPID','--value'])
|
|
require(code==0 and main.strip()==str(pid) and (uid,gid)==(0,0) and current['uids']==[0]*4 and
|
|
current['group']=='/sys/fs/cgroup/system.slice/archipelago-d09-lifecycle.service',
|
|
'generation actor must be current root D09 MainPID')
|
|
mutating=request.get('method')!='inspect'
|
|
require(generation_pending is None and (not mutating or (leases.supervisor is None and leases.retiring is None)),
|
|
'generation handler busy or management lifetime not retired')
|
|
if mutating:
|
|
leases.generation+=1 # old checker results cannot poison this transition
|
|
send(generation_worker.channel,request)
|
|
generation_pending=(channel,mutating)
|
|
else:
|
|
leases.accept(channel, request)
|
|
except (OSError, ValueError, KeyError, TypeError):
|
|
if channel in leases.clients:
|
|
raise # An uncertain delivered grant fences the whole owner.
|
|
channel.close()
|
|
del pending[channel]
|
|
finally:
|
|
if generation_pending is not None:
|
|
generation_pending[0].close()
|
|
leases.stop()
|
|
for channel in pending:
|
|
channel.close()
|
|
if listener is not None:
|
|
listener.close()
|
|
children.request_stop()
|
|
# The journal remains owner-active if storage stalls here. Either state
|
|
# refuses a restart. Do NOT perform blocking journal I/O in this observer.
|
|
|
|
|
|
def resume_retired_layout(policy,journal):
|
|
require(policy.value['version']==3,'explicit layout migration required')
|
|
old,_=policy_module.json_read(journal.path)
|
|
require(old['state']=='writers-retired' and old['configuration_sha256']==policy.digest and old['boot']==journal.boot,
|
|
'no same-boot authenticated retired-writer checkpoint')
|
|
writers=old['evidence']['writers']
|
|
expected={(int(pid),start,spec['path']) for spec in policy.value['cohorts'] for pid,start in spec['members'].items()}
|
|
require(len(writers)==len(expected) and all(any(item['pid']==pid and item['start']==start and
|
|
Path(item['group']).is_relative_to(Path(group)) for item in writers) for pid,start,group in expected),
|
|
'retired-writer checkpoint differs from approved inventory')
|
|
policy.verify_files();policy.verify_trees();policy.preserved();semantic_checks(policy);ssh_checks(policy)
|
|
managed={a['uid'] for a in policy.value['accounts'].values()}
|
|
require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),'migration writers remain')
|
|
for spec in policy.value['cohorts']:kernel.replacement_tree(spec,policy.runtime)
|
|
import c01_migration_layout
|
|
layout=c01_migration_layout.apply_retired(policy,resume=True)
|
|
copies=kernel.quarantine(policy,resume=True)
|
|
kernel.verify_private(policy)
|
|
require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),'migration writer appeared')
|
|
return {'writers':writers,'copies':copies,'layout':layout,'operations':OPERATIONS}
|
|
|
|
|
|
def transition(policy, journal, children, resume=False):
|
|
require(resume or not journal.path.exists(), 'prior access transition/lifetime requires offline reconciliation')
|
|
def body(control):
|
|
try:
|
|
send(control, {'progress': 'preflight-started'})
|
|
if resume:
|
|
evidence=resume_retired_layout(policy,journal)
|
|
else:
|
|
preflight(policy) # Deterministic conflicts leave no pending migration.
|
|
journal.begin_transition()
|
|
send(control, {'progress': 'intent-durable'})
|
|
evidence = retirement(policy, lambda step: send(control, {'progress': step}),journal.record)
|
|
except (OSError, ValueError) as error:
|
|
# Fixed root-only checks contain no credential contents. Retain a
|
|
# bounded diagnostic instead of losing the first failure as EOF.
|
|
# The pending journal and all retirement obligations remain intact.
|
|
import traceback
|
|
frame = traceback.extract_tb(error.__traceback__)[-1]
|
|
location = f'{Path(frame.filename).name}:{frame.lineno}:{frame.name}'
|
|
send(control, {'failure': type(error).__name__, 'reason': (location + ': ' + str(error))[:256]})
|
|
require(control.recv(8) == b'RELEASE', 'failed transition release refused')
|
|
raise
|
|
# Potentially large process inventory stays in the protected journal;
|
|
# never fit/truncate it into a protocol packet or log secret contents.
|
|
journal.record('operations-verified', evidence)
|
|
send(control, {'complete': True})
|
|
# Keep the direct child alive until MainPID consumes the completion.
|
|
require(control.recv(8) == b'RELEASE', 'transition release refused')
|
|
record = children.spawn('collector', body, keep=(journal.fd,))
|
|
deadline = kernel.now() + 30_000_000_000
|
|
completed = False
|
|
while not children.stopping:
|
|
children.poll()
|
|
require('collector' in children.children, 'transition helper exited before completion')
|
|
require(kernel.now() < deadline, 'transition deadline expired; retirement remains fenced')
|
|
ready, _, _ = select.select([record.channel], [], [], .05)
|
|
if ready:
|
|
packet = receive(record.channel, (record.pid, 0, 0))
|
|
if set(packet) == {'failure', 'reason'}:
|
|
require(packet['failure'] in ('OSError', 'PermissionError', 'FileNotFoundError',
|
|
'ProcessLookupError', 'ValueError') and type(packet['reason']) is str and
|
|
len(packet['reason']) <= 256, 'invalid transition diagnostic')
|
|
record.channel.send(b'RELEASE')
|
|
raise ValueError('access transition refused: ' + packet['reason'])
|
|
if packet == {'complete': True}:
|
|
record.channel.send(b'RELEASE')
|
|
completed = True
|
|
break
|
|
require(set(packet) == {'progress'} and packet['progress'] in {
|
|
'preflight-started', 'intent-durable',
|
|
'frozen-inventoried-cohorts', 'masked-and-stopped-activation', 'installed-exact-access-policy',
|
|
'reloaded-ssh-retired-polkit-cache', 'kernel-observed-old-writer-exit'}, 'transition progress refused')
|
|
print('C01_ACCESS_TRANSITION: ' + packet['progress'], flush=True)
|
|
require(completed, 'transition interrupted')
|
|
while children.children:
|
|
children.poll()
|
|
require(not children.stopping and kernel.now() < deadline, 'transition helper retirement unresolved')
|
|
time.sleep(.02)
|
|
require(children.results.get('collector') == 0, 'transition helper failed')
|
|
value, _ = policy_module.json_read(journal.path)
|
|
require(value['state'] == 'operations-verified', 'missing verified transition operations')
|
|
journal.record('retired', value['evidence'])
|
|
|
|
|
|
def retire_endpoint(evidence):
|
|
"""Offline only, after all former lifetimes retire under the persistent lock."""
|
|
policy_module.protected(BASE, directory=True)
|
|
if not os.path.lexists(SOCKET):
|
|
return
|
|
meta = SOCKET.lstat()
|
|
require(stat.S_ISSOCK(meta.st_mode) and (meta.st_uid, meta.st_gid) == (0, 0) and
|
|
stat.S_IMODE(meta.st_mode) == 0o600 and meta.st_nlink == 1 and
|
|
policy_module.identity(meta) == evidence.get('endpoint'), 'unowned stale access endpoint')
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET | socket.SOCK_NONBLOCK) as probe:
|
|
try:
|
|
probe.connect(str(SOCKET))
|
|
except OSError as error:
|
|
require(error.errno == errno.ECONNREFUSED, 'access endpoint retirement not proved')
|
|
else:
|
|
raise ValueError('former access listener remains active')
|
|
current = SOCKET.lstat()
|
|
require(policy_module.identity(current) == policy_module.identity(meta) and stat.S_ISSOCK(current.st_mode) and
|
|
current.st_nlink == 1 and (current.st_mtime_ns, current.st_ctime_ns) ==
|
|
(meta.st_mtime_ns, meta.st_ctime_ns), 'stale access endpoint changed')
|
|
SOCKET.unlink()
|
|
|
|
|
|
def requalify(policy, journal, children):
|
|
"""Offline, fully quiescent owner restart; never a remote reset/unfence RPC.
|
|
|
|
A partial migration cannot use this path. Existing runtime apps must already
|
|
be stopped by their owning coordinator; this operation kills no process and
|
|
does not restore old credentials, start apps, or change desired-state data.
|
|
"""
|
|
previous, _ = policy_module.json_read(journal.path)
|
|
completed_history(previous, policy)
|
|
require(previous['state'] == 'owner-active' or previous['boot'] != journal.boot,
|
|
'same-boot completed retirement already eligible')
|
|
prior = previous['owner']
|
|
prior_group = policy_module.canonical(prior['group'])
|
|
require(prior_group.is_relative_to(kernel.CGROUP) and prior_group.name == 'supervisor',
|
|
'invalid former owner hierarchy')
|
|
if previous['boot'] == journal.boot:
|
|
try:
|
|
item = kernel.process(prior['pid'])
|
|
except (FileNotFoundError, ProcessLookupError):
|
|
pass
|
|
else:
|
|
require(item['start'] != prior['start'], 'former access owner remains alive')
|
|
if prior_group.parent.exists():
|
|
require(not any((prior_group.parent / ('c01-' + role)).exists() for role in children_module.ROLES) and
|
|
kernel.members(prior_group.parent) <= {os.getpid()}, 'former access helper retirement unresolved')
|
|
for field in ('broker_cgroup', 'supervisor_cgroup'):
|
|
group = Path(policy.value[field])
|
|
require(not group.exists() or (not kernel.members(group) and
|
|
not any((group / ('c01-' + role)).exists() for role in children_module.ROLES)),
|
|
'consumer retirement unresolved')
|
|
managed = {a['uid'] for a in policy.value['accounts'].values()}
|
|
require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),
|
|
'offline requalification requires quiescent managed/runtime principals')
|
|
# Fence before I/O helper work; interruption needs explicit offline review,
|
|
# never an automatic retry that trusts the prior journal state.
|
|
journal.record('requalification-pending', previous['evidence'])
|
|
def body(control):
|
|
check_current(policy, None)
|
|
require(all(not managed.intersection(item['uids']) for item in kernel.all_processes()),
|
|
'managed process appeared during requalification')
|
|
send(control, {'verified': True})
|
|
require(control.recv(8) == b'RELEASE', 'requalification release refused')
|
|
record = children.spawn('collector', body)
|
|
deadline = kernel.now() + 30_000_000_000
|
|
while True:
|
|
children.poll()
|
|
require(not children.stopping and 'collector' in children.children and kernel.now() < deadline,
|
|
'requalification helper failed/stalled')
|
|
if select.select([record.channel], [], [], .05)[0]:
|
|
require(receive(record.channel, (record.pid, 0, 0)) == {'verified': True}, 'requalification evidence refused')
|
|
record.channel.send(b'RELEASE')
|
|
break
|
|
while children.children:
|
|
children.poll()
|
|
require(not children.stopping and kernel.now() < deadline, 'requalification helper retirement unresolved')
|
|
time.sleep(.02)
|
|
require(children.results.get('collector') == 0, 'requalification helper failed')
|
|
retire_endpoint(previous['evidence'])
|
|
evidence = {**previous['evidence'], 'requalification': {'prior_boot': previous['boot'],
|
|
'former_owner': prior, 'operation': 'quiescent-kernel-and-current-policy-reverified'}}
|
|
journal.record('retired', evidence)
|
|
|
|
|
|
def main():
|
|
require(len(sys.argv) == 2 and sys.argv[1] in ('serve', 'transition', 'requalify', 'resume-layout'), 'fixed access operation required')
|
|
require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0) and
|
|
len(list(Path('/proc/self/task').iterdir())) == 1, 'single-threaded root owner required')
|
|
os.umask(0o077)
|
|
require(children_module.LIBC.prctl(4, 0, 0, 0, 0) == 0, 'owner dumpability protection unavailable')
|
|
policy = policy_module.Policy()
|
|
journal = Journal(policy)
|
|
children = None
|
|
try:
|
|
children = children_module.Children()
|
|
require(not any((children.root / ('c01-' + role)).exists() for role in children_module.ROLES),
|
|
'prior access helper cohort remains unresolved')
|
|
# A crash can leave a prior checker/transition helper (including D-state)
|
|
# behind. Children never adopts an existing role group. Journal blocks too.
|
|
if sys.argv[1] == 'transition':
|
|
transition(policy, journal, children)
|
|
elif sys.argv[1] == 'resume-layout':
|
|
transition(policy,journal,children,resume=True)
|
|
elif sys.argv[1] == 'requalify':
|
|
requalify(policy, journal, children)
|
|
else:
|
|
serve(policy, journal, children)
|
|
finally:
|
|
if children is not None:
|
|
children.request_stop()
|
|
while children.children or children.reserved:
|
|
children.poll()
|
|
time.sleep(.05)
|
|
children.close()
|
|
journal.close() # Only after owned helper/descendants really retire.
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|