744 lines
34 KiB
Python
744 lines
34 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent root application-pair owner. No app/package/OS operations.
|
|
|
|
Production is bundled with public_update_prepare.py by the installed binary.
|
|
Only a freshly authenticated immutable snapshot may enter submit(). Recovery
|
|
uses root-owned preimages and a write-ahead journal, never legacy app backups.
|
|
"""
|
|
import contextlib
|
|
import shutil
|
|
import time
|
|
import urllib.request
|
|
|
|
TRANSACTIONS = Path('/var/lib/archipelago-public-transactions')
|
|
BINARY = Path('/usr/local/bin/archipelago')
|
|
FRONTEND = Path('/opt/archipelago/web-ui')
|
|
UNIT_DIR = Path('/etc/systemd/system')
|
|
DROPIN = UNIT_DIR / 'archipelago.service.d/70-public-application-update.conf'
|
|
OWNER_DROPINS = (DROPIN, *(UNIT_DIR / (unit + '.d/70-public-application-update.conf')
|
|
for unit in MAINTENANCE_UNITS))
|
|
BOOT_WANTS = UNIT_DIR / 'basic.target.wants'
|
|
TERMINAL = {'committed', 'restored', 'aborted'}
|
|
ENV = {'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'HOME': '/nonexistent', 'LANG': 'C'}
|
|
|
|
|
|
def command(*args, timeout=90):
|
|
return subprocess.check_output(args, env=ENV, cwd='/', timeout=timeout, text=True)
|
|
|
|
|
|
def atomic(path, data, mode=0o600):
|
|
temporary = path.with_name('.' + path.name + '-' + uuid.uuid4().hex)
|
|
with temporary.open('xb') as stream:
|
|
stream.write(data)
|
|
stream.flush()
|
|
os.fchmod(stream.fileno(), mode)
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, path)
|
|
sync_directory(path.parent)
|
|
|
|
|
|
def journal(tx, value, phase):
|
|
value = dict(value, phase=phase)
|
|
atomic(tx / 'state.json', json.dumps(value, sort_keys=True).encode())
|
|
return value
|
|
|
|
|
|
def secure_read(path, limit=MAX_DOCUMENT):
|
|
fd = trusted_open(path)
|
|
with os.fdopen(fd, 'rb') as stream:
|
|
raw = stream.read(limit + 1)
|
|
if len(raw) > limit:
|
|
raise ValueError('oversized protected record')
|
|
return raw
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def owner_lock():
|
|
os.close(trusted_open(TRANSACTIONS.parent, directory=True))
|
|
TRANSACTIONS.mkdir(mode=0o755, exist_ok=True)
|
|
os.close(trusted_open(TRANSACTIONS, directory=True))
|
|
fd = os.open(TRANSACTIONS / 'lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_uid or info.st_mode & 0o077:
|
|
raise ValueError('untrusted worker lock')
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
yield
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def load_transaction(name):
|
|
if not re.fullmatch('[0-9a-f]{32}', name):
|
|
raise ValueError('invalid transaction pointer')
|
|
tx = TRANSACTIONS / name
|
|
state = json.loads(secure_read(tx / 'state.json'), object_pairs_hook=pairs)
|
|
if state.get('schema') != 1 or state.get('id') != name:
|
|
raise ValueError('invalid transaction state')
|
|
return tx, state
|
|
|
|
|
|
def current():
|
|
try:
|
|
name = secure_read(TRANSACTIONS / 'current').decode()
|
|
except FileNotFoundError:
|
|
return None, None
|
|
return load_transaction(name)
|
|
|
|
|
|
def check_owner_dropins(previous, incoming=None):
|
|
for path in OWNER_DROPINS:
|
|
allowed = [owner_dropin(value, path) for value in (previous, incoming) if value]
|
|
if os.path.lexists(path) and secure_read(path) not in allowed:
|
|
raise ValueError('foreign public-update service drop-in')
|
|
|
|
|
|
def owner_dropin(state, path):
|
|
value = state['dropin']
|
|
if state.get('runtime_startup_guard'):
|
|
if path == DROPIN:
|
|
# Historical binaries unconditionally repromote the frontend's
|
|
# runtime on every startup, deleting locally retained app sources.
|
|
# Hide only that bootstrap input inside the manager namespace. The
|
|
# host owner and nginx still see the exact authenticated frontend.
|
|
value += '[Service]\nInaccessiblePaths=-/opt/archipelago/web-ui/archipelago-runtime\n'
|
|
else:
|
|
value += 'ConditionPathExists=!/var/lib/archipelago-public-transactions/maintenance-paused\n'
|
|
return value.encode()
|
|
|
|
|
|
def install_owner_dropins(state, *, reload=True):
|
|
previous = load_transaction(state['previous_id'])[1] if state.get('previous_id') else None
|
|
check_owner_dropins(previous, state)
|
|
for path in OWNER_DROPINS:
|
|
path.parent.mkdir(mode=0o755, exist_ok=True)
|
|
os.close(trusted_open(path.parent, directory=True))
|
|
atomic(path, owner_dropin(state, path), 0o644)
|
|
if reload:
|
|
command('/usr/bin/systemctl', 'daemon-reload')
|
|
|
|
|
|
def retain_boot_recovery(unit):
|
|
BOOT_WANTS.mkdir(mode=0o755, exist_ok=True)
|
|
os.close(trusted_open(BOOT_WANTS, directory=True))
|
|
link = BOOT_WANTS / unit
|
|
target = str(UNIT_DIR / unit)
|
|
if os.path.lexists(link):
|
|
info = link.lstat()
|
|
if not stat.S_ISLNK(info.st_mode) or info.st_uid or os.readlink(link) != target:
|
|
raise ValueError('foreign public recovery boot link')
|
|
else:
|
|
link.symlink_to(target)
|
|
sync_directory(BOOT_WANTS)
|
|
|
|
|
|
|
|
def digest_file(path):
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
with os.fdopen(fd, 'rb') as stream:
|
|
if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
|
|
raise ValueError('nonregular code file')
|
|
return hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
|
|
|
|
def copy_file_fd(fd, destination, executable=False):
|
|
before = os.fstat(fd)
|
|
if not stat.S_ISREG(before.st_mode) or before.st_size > MAX_COMPONENT:
|
|
raise ValueError('invalid preimage file')
|
|
digest = hashlib.sha256()
|
|
with os.fdopen(os.dup(fd), 'rb') as source, destination.open('xb') as target:
|
|
copied = 0
|
|
while block := source.read(1024 * 1024):
|
|
copied += len(block)
|
|
if copied > before.st_size:
|
|
raise ValueError('preimage grew during snapshot')
|
|
target.write(block)
|
|
digest.update(block)
|
|
target.flush()
|
|
os.fchmod(target.fileno(), 0o555 if executable else 0o444)
|
|
os.fsync(target.fileno())
|
|
after = os.fstat(fd)
|
|
if copied != before.st_size or (before.st_dev, before.st_ino, before.st_size,
|
|
before.st_mtime_ns, before.st_ctime_ns) != (after.st_dev, after.st_ino,
|
|
after.st_size, after.st_mtime_ns, after.st_ctime_ns):
|
|
raise ValueError('preimage mutated during snapshot')
|
|
return digest.hexdigest()
|
|
|
|
|
|
def durable_mode(path, mode):
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
try:
|
|
os.fchmod(fd, mode)
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def tree(path, destination=None, with_size=False):
|
|
"""No-follow fd walk; snapshot or identify code, including executable bits.
|
|
|
|
Never follows app-controlled links or trusts path-based copytree. A changing
|
|
directory/file refuses, and partial copies are retained as failure evidence.
|
|
"""
|
|
entries = []
|
|
size = 0
|
|
|
|
def walk(fd, prefix, out):
|
|
nonlocal size
|
|
before = os.fstat(fd)
|
|
for name in sorted(os.listdir(fd)):
|
|
child = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd)
|
|
try:
|
|
info = os.fstat(child)
|
|
relative = prefix + name
|
|
if stat.S_ISDIR(info.st_mode):
|
|
entries.append([relative, 'directory'])
|
|
target = out / name if out else None
|
|
if target:
|
|
target.mkdir(mode=0o700)
|
|
walk(child, relative + '/', target)
|
|
if target:
|
|
target.chmod(0o555)
|
|
sync_directory(target)
|
|
elif stat.S_ISREG(info.st_mode) and not info.st_mode & 0o6000:
|
|
size += info.st_size
|
|
executable = bool(info.st_mode & 0o111)
|
|
if out:
|
|
digest = copy_file_fd(child, out / name, executable)
|
|
else:
|
|
with os.fdopen(os.dup(child), 'rb') as source:
|
|
digest = hashlib.file_digest(source, 'sha256').hexdigest()
|
|
entries.append([relative, 'file', executable, info.st_size, digest])
|
|
else:
|
|
raise ValueError('unsupported frontend preimage inode')
|
|
if len(entries) > 20000 or size > 4 * 1024**3:
|
|
raise ValueError('frontend preimage exceeds bounds')
|
|
finally:
|
|
os.close(child)
|
|
after = os.fstat(fd)
|
|
if (before.st_mtime_ns, before.st_ctime_ns) != (after.st_mtime_ns, after.st_ctime_ns):
|
|
raise ValueError('frontend directory mutated during snapshot')
|
|
|
|
fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
if destination:
|
|
destination.mkdir(mode=0o700)
|
|
walk(fd, '', destination)
|
|
if destination:
|
|
destination.chmod(0o555)
|
|
sync_directory(destination)
|
|
finally:
|
|
os.close(fd)
|
|
digest = hashlib.sha256(json.dumps(entries, separators=(',', ':')).encode()).hexdigest()
|
|
return (digest, size + 4096 * len(entries)) if with_size else digest
|
|
|
|
|
|
def extract(archive_path, destination, require_declaration=True):
|
|
validate_archive(archive_path, require_declaration)
|
|
destination.mkdir(mode=0o700)
|
|
with tarfile.open(archive_path, 'r:gz') as archive:
|
|
for member in archive:
|
|
name = member.name.removeprefix('./').rstrip('/')
|
|
if name in ('', '.'):
|
|
continue
|
|
target = destination / name
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
if member.isdir():
|
|
target.mkdir(exist_ok=True)
|
|
else:
|
|
with archive.extractfile(member) as source, target.open('xb') as output:
|
|
shutil.copyfileobj(source, output, 1024 * 1024)
|
|
output.flush()
|
|
os.fchmod(output.fileno(), 0o555 if member.mode & 0o111 else 0o444)
|
|
os.fsync(output.fileno())
|
|
for directory, _, _ in os.walk(destination, topdown=False):
|
|
Path(directory).chmod(0o555)
|
|
sync_directory(Path(directory))
|
|
return tree(destination)
|
|
|
|
|
|
def pair_identity(binary, frontend):
|
|
return {'binary': digest_file(binary), 'frontend': tree(frontend)}
|
|
|
|
|
|
def require_space(tx):
|
|
# Budget both forward and restore copies without deleting retired evidence.
|
|
# Sum requirements for shared filesystems rather than counting their free
|
|
# space independently. No reservation is inferred from app-writable state.
|
|
old_frontend = tree(FRONTEND, with_size=True)[1]
|
|
new_frontend = tree(tx / 'new-frontend', with_size=True)[1]
|
|
budgets = {}
|
|
for path, amount in [(tx, BINARY.stat().st_size + 2 * old_frontend),
|
|
(FRONTEND.parent, old_frontend + new_frontend),
|
|
(BINARY.parent, BINARY.stat().st_size + (tx / 'new-binary').stat().st_size)]:
|
|
device = path.stat().st_dev
|
|
previous = budgets.get(device, (path, 256 * 1024**2))
|
|
budgets[device] = (path, previous[1] + amount)
|
|
if json.loads(secure_read(tx / 'signed-manifest.json')).get('application_update') == RUNTIME_POLICY:
|
|
for destination, relative, kind, required in RUNTIME_OBJECTS.values():
|
|
old = Path(destination)
|
|
new = tx / 'new-frontend/archipelago-runtime' / relative if relative else None
|
|
def size(path):
|
|
if path is None or not os.path.lexists(path):
|
|
return 0
|
|
identity = runtime_identity(path)
|
|
return tree(path, with_size=True)[1] if identity['kind'] == 'directory' else path.stat().st_size + 4096
|
|
old_size, new_size = size(old), size(new)
|
|
parent = old.parent
|
|
while not parent.exists():
|
|
parent = parent.parent
|
|
for path, amount in ((tx, old_size + new_size), (parent, old_size + new_size)):
|
|
device = path.stat().st_dev
|
|
previous = budgets.get(device, (path, 256 * 1024**2))
|
|
budgets[device] = (path, previous[1] + amount)
|
|
for path, required in budgets.values():
|
|
free = os.statvfs(path)
|
|
if free.f_bavail * free.f_frsize < required:
|
|
raise ValueError('insufficient space for protected preimages, installation and recovery')
|
|
|
|
|
|
def check_preimages(tx, state):
|
|
for label in ('old', 'new'):
|
|
if pair_identity(tx / (label + '-binary'), tx / (label + '-frontend')) != state[label]:
|
|
raise ValueError('protected pair preimage identity changed')
|
|
|
|
|
|
def custody_frontend_parent():
|
|
os.close(trusted_open(FRONTEND.parent.parent, directory=True))
|
|
fd = os.open(FRONTEND.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
info = os.fstat(fd)
|
|
import pwd
|
|
if info.st_uid not in (0, pwd.getpwnam('archipelago').pw_uid) or info.st_mode & 0o002:
|
|
raise ValueError('nonstandard application prefix owner')
|
|
# Only the code prefix directory, never its children or any app data.
|
|
# Root custody prevents the service user renaming our installation paths.
|
|
os.fchown(fd, 0, 0)
|
|
os.fchmod(fd, 0o755)
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def replace_pair(tx, state, label, *, unprepared_adoption=False):
|
|
if unprepared_adoption:
|
|
if (not state.get('adoption') or label != 'old' or state['phase'] not in ('queued', 'preparing')
|
|
or pair_identity(tx / 'old-binary', tx / 'old-frontend') != state['old']):
|
|
raise ValueError('first-generation abort lacks authenticated old preimages')
|
|
# Derived candidate staging may be incomplete. It is never a source for
|
|
# this restore; both restored objects are the authenticated old pair.
|
|
else:
|
|
check_preimages(tx, state)
|
|
os.close(trusted_open(BINARY.parent, directory=True))
|
|
os.close(trusted_open(FRONTEND.parent, directory=True))
|
|
# Each target may be absent at an interrupted directory rename, or match
|
|
# either transaction image. Foreign modifications are never overwritten.
|
|
for key, target, identify in [('binary', BINARY, digest_file), ('frontend', FRONTEND, tree)]:
|
|
allowed = [state['old'][key], state['new'][key]]
|
|
if state.get('adoption'):
|
|
allowed.append(state['initial'][key])
|
|
if os.path.lexists(target) and identify(target) not in allowed:
|
|
raise ValueError('foreign installed code edit; recovery requires operator review')
|
|
for key, target in [('frontend', FRONTEND), ('binary', BINARY)]:
|
|
temporary = target.with_name('.public-new-' + uuid.uuid4().hex)
|
|
if key == 'frontend':
|
|
tree(tx / (label + '-frontend'), temporary)
|
|
# Normalize readable deployed modes; preserve executable bits.
|
|
for directory, _, files in os.walk(temporary):
|
|
for name in files:
|
|
p = Path(directory) / name
|
|
durable_mode(p, 0o755 if p.stat().st_mode & 0o111 else 0o644)
|
|
durable_mode(Path(directory), 0o755)
|
|
if os.path.lexists(target):
|
|
target.rename(target.with_name('.public-retired-' + uuid.uuid4().hex))
|
|
sync_directory(target.parent)
|
|
else:
|
|
fd = trusted_open(tx / (label + '-binary'))
|
|
try:
|
|
copy_file_fd(fd, temporary, True)
|
|
finally:
|
|
os.close(fd)
|
|
durable_mode(temporary, 0o755)
|
|
os.replace(temporary, target)
|
|
sync_directory(target.parent)
|
|
if pair_identity(BINARY, FRONTEND) != state[label]:
|
|
raise ValueError('installed pair identity mismatch')
|
|
|
|
|
|
def manager_active():
|
|
return command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '--value').strip() == 'active'
|
|
|
|
|
|
def manager_adoptable():
|
|
# Type=notify remains activating until the initial migration hands off.
|
|
# Bind that starting process to the actual installed executable inode.
|
|
fields = command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '-p', 'MainPID')
|
|
properties = dict(line.split('=', 1) for line in fields.splitlines() if '=' in line)
|
|
if properties.get('ActiveState') not in ('active', 'activating'):
|
|
return False
|
|
pid = int(properties.get('MainPID', '0'))
|
|
return pid > 0 and os.path.samefile(f'/proc/{pid}/exe', BINARY)
|
|
|
|
|
|
def stop_manager():
|
|
command('/usr/bin/systemctl', 'stop', 'archipelago.service', timeout=180)
|
|
if manager_active():
|
|
raise ValueError('manager did not stop')
|
|
|
|
|
|
def verify_running(tx, state, label):
|
|
deadline = time.monotonic() + 180
|
|
last = None
|
|
while time.monotonic() < deadline:
|
|
if command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'ActiveState', '--value').strip() == 'failed':
|
|
raise RuntimeError('manager failed after pair replacement')
|
|
try:
|
|
if not manager_active():
|
|
raise ValueError('manager not active')
|
|
pid = int(command('/usr/bin/systemctl', 'show', 'archipelago.service', '-p', 'MainPID', '--value'))
|
|
# Hash the actually executing inode, not a version string or stale
|
|
# socket response. /proc/PID/exe is deliberately followed here.
|
|
with open(f'/proc/{pid}/exe', 'rb') as stream:
|
|
actual = hashlib.file_digest(stream, 'sha256').hexdigest()
|
|
if actual != state[label]['binary']:
|
|
raise ValueError('manager executing a different binary')
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
with opener.open('http://127.0.0.1:5678/health', timeout=3) as response:
|
|
health = json.loads(response.read(65537))
|
|
if health.get('version') != state[label + '_version']:
|
|
raise RuntimeError('running backend version differs from the admitted pair')
|
|
if health.get('status') != 'ok':
|
|
raise ValueError('backend health not ready')
|
|
with opener.open('http://127.0.0.1/', timeout=3) as response:
|
|
index = response.read(8 * 1024**2 + 1)
|
|
expected_frontend = tx / (label + '-frontend') if tx else FRONTEND
|
|
if hashlib.sha256(index).hexdigest() != digest_file(expected_frontend / 'index.html'):
|
|
raise ValueError('served frontend differs from installed pair')
|
|
if pair_identity(BINARY, FRONTEND) != state[label]:
|
|
raise ValueError('installed pair changed during verification')
|
|
if state.get('runtime'):
|
|
runtime_verify_installed(state, label)
|
|
return
|
|
except (OSError, ValueError, subprocess.SubprocessError) as error:
|
|
if str(last) != str(error):
|
|
state['last_probe_error'] = str(error)
|
|
if tx:
|
|
journal(tx, state, state['phase'])
|
|
last = error
|
|
time.sleep(1)
|
|
raise ValueError(f'post-restart pair verification failed: {last}')
|
|
|
|
|
|
def recovery(tx, state, boot=False):
|
|
require_legacy_service()
|
|
if state['phase'] in ('queued', 'preparing'):
|
|
if state.get('adoption'):
|
|
# The historical installer replaced its pair before this owner
|
|
# existed. Abort this first-generation handoff to its authenticated
|
|
# old pair even if runtime preparation never became durable.
|
|
if not boot:
|
|
stop_manager()
|
|
command('/usr/bin/systemctl', 'daemon-reload')
|
|
replace_pair(tx, state, 'old', unprepared_adoption=True)
|
|
retain_legacy_marker(tx, state)
|
|
state = journal(tx, state, 'restored-pending-start')
|
|
else:
|
|
# Normal updates cannot mutate code before prepared is durable.
|
|
state = journal(tx, state, 'aborted')
|
|
elif state['phase'] not in ('restored', 'aborted'):
|
|
state = journal(tx, state, 'recovering')
|
|
if not boot:
|
|
stop_manager()
|
|
check_preimages(tx, state)
|
|
if state.get('runtime'):
|
|
runtime_hold_maintenance(state)
|
|
runtime_replace(tx, state, 'old')
|
|
replace_pair(tx, state, 'old')
|
|
retain_legacy_marker(tx, state)
|
|
state = journal(tx, state, 'restored-pending-start')
|
|
if not boot and state['was_active']:
|
|
command('/usr/bin/systemctl', 'start', 'archipelago.service', timeout=180)
|
|
if state['phase'] == 'restored-pending-start':
|
|
verify_running(tx, state, 'old')
|
|
runtime_resume_units(state)
|
|
state = journal(tx, state, 'restored')
|
|
elif state['phase'] == 'aborted':
|
|
runtime_resume_units(state)
|
|
elif boot and state['phase'] == 'restored-pending-start':
|
|
command('/usr/bin/systemctl', 'start', '--no-block', state['worker_unit'])
|
|
return state
|
|
|
|
|
|
def run_worker(expected_id=None):
|
|
with owner_lock():
|
|
tx, state = current()
|
|
if state and expected_id is not None and state['id'] != expected_id:
|
|
return
|
|
if not state or state['phase'] in TERMINAL:
|
|
return
|
|
if state['phase'] != 'queued' or state.get('recovery_requested'):
|
|
try:
|
|
recovery(tx, state)
|
|
except Exception as error:
|
|
tx, state = current()
|
|
journal(tx, dict(state, failure=str(error)), state['phase'])
|
|
raise
|
|
return
|
|
try:
|
|
require_legacy_service()
|
|
install_owner_dropins(state, reload=not state.get('adoption'))
|
|
if not (manager_adoptable() if state.get('adoption') else manager_active()):
|
|
if state.get('adoption'):
|
|
recovery(tx, state)
|
|
else:
|
|
journal(tx, dict(state, was_active=False), 'aborted')
|
|
return
|
|
state = journal(tx, state, 'preparing')
|
|
custody_frontend_parent()
|
|
baseline = pair_identity(BINARY, FRONTEND)
|
|
if state.get('adoption'):
|
|
if baseline != state['initial']:
|
|
raise ValueError('first-generation installed code changed before admission')
|
|
check_preimages(tx, state)
|
|
else:
|
|
state['old'] = baseline
|
|
verify_running(None, state, 'old')
|
|
require_space(tx)
|
|
stop_manager()
|
|
if state.get('adoption'):
|
|
# Reload only after the first manager has stopped. Reloading
|
|
# file-backed units discards its EXTEND_TIMEOUT_USEC deadline.
|
|
command('/usr/bin/systemctl', 'daemon-reload')
|
|
if state.get('runtime_requested'):
|
|
state = runtime_pause_units(tx, state)
|
|
if not state.get('adoption'):
|
|
fd = trusted_open(BINARY)
|
|
try:
|
|
copy_file_fd(fd, tx / 'old-binary', True)
|
|
finally:
|
|
os.close(fd)
|
|
tree(FRONTEND, tx / 'old-frontend')
|
|
state['old'] = pair_identity(tx / 'old-binary', tx / 'old-frontend')
|
|
if pair_identity(BINARY, FRONTEND) != baseline:
|
|
raise ValueError('installed code changed after baseline verification')
|
|
# Preserve separately installed UI extras only if not superseded by
|
|
# the signed archive. Copies originate from the frozen old tree.
|
|
for name in ('aiui', 'archipelago-companion.apk'):
|
|
old, new = tx / 'old-frontend' / name, tx / 'new-frontend' / name
|
|
if old.exists() and not new.exists():
|
|
if old.is_dir():
|
|
tree(old, new)
|
|
else:
|
|
fd = trusted_open(old)
|
|
try:
|
|
copy_file_fd(fd, new, bool(old.stat().st_mode & 0o111))
|
|
finally:
|
|
os.close(fd)
|
|
state['new'] = pair_identity(tx / 'new-binary', tx / 'new-frontend')
|
|
if state.get('runtime_requested'):
|
|
state = runtime_prepare(tx, state)
|
|
sync_directory(tx / 'new-frontend')
|
|
sync_directory(tx)
|
|
state = journal(tx, state, 'prepared')
|
|
require_legacy_service()
|
|
state = journal(tx, state, 'installing')
|
|
check_preimages(tx, state)
|
|
if state.get('runtime'):
|
|
runtime_replace(tx, state, 'new')
|
|
replace_pair(tx, state, 'new')
|
|
state = journal(tx, state, 'verifying')
|
|
command('/usr/bin/systemctl', 'start', 'archipelago.service', timeout=180)
|
|
verify_running(tx, state, 'new')
|
|
retain_legacy_marker(tx, state)
|
|
runtime_resume_units(state)
|
|
journal(tx, state, 'committed')
|
|
except Exception as error:
|
|
# Preserve the exact cause even when recovery itself fails. A killed
|
|
# worker is retried by PID1; a reboot invokes recovery before manager.
|
|
tx, state = current()
|
|
state['failure'] = str(error)
|
|
state = journal(tx, state, state['phase'])
|
|
recovery(tx, state)
|
|
raise
|
|
|
|
|
|
def submit(identity, bundle):
|
|
if not re.fullmatch('[0-9a-f]{64}', identity):
|
|
raise ValueError('invalid prepared identity')
|
|
require_legacy_service()
|
|
ready = SNAPSHOTS / identity
|
|
raw = secure_read(ready / 'signed-manifest.json')
|
|
document = json.loads(raw, object_pairs_hook=pairs)
|
|
old_version = authenticate(raw)
|
|
components = validate_document(document, old_version)
|
|
expected = hashlib.sha256(json.dumps([document['version'], sorted([
|
|
[c['name'], c['new_version'], c['sha256'], c['size_bytes']] for c in components])],
|
|
separators=(',', ':'), ensure_ascii=True).encode()).hexdigest()
|
|
if expected != identity:
|
|
raise ValueError('prepared identity differs from authenticated release')
|
|
with owner_lock():
|
|
previous_tx, previous = current()
|
|
if previous and previous['phase'] not in TERMINAL:
|
|
raise ValueError('public application transaction already pending')
|
|
require_legacy_service()
|
|
if os.path.lexists(STAGING.parent / 'update-pending-verify.json'):
|
|
raise ValueError('unresolved historical OTA verification marker')
|
|
if not manager_active():
|
|
raise ValueError('only an active standard manager may request activation')
|
|
check_owner_dropins(previous)
|
|
tx = TRANSACTIONS / uuid.uuid4().hex
|
|
tx.mkdir(mode=0o700)
|
|
for component in components:
|
|
os.close(trusted_open(ready / component['name']))
|
|
source = os.open(ready, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
for component in components:
|
|
copy_component(source, tx / component['name'], component)
|
|
finally:
|
|
os.close(source)
|
|
(tx / 'archipelago').rename(tx / 'new-binary')
|
|
durable_mode(tx / 'new-binary', 0o555)
|
|
extract(tx / f"archipelago-frontend-{document['version']}.tar.gz", tx / 'new-frontend')
|
|
write_manifest(tx / 'signed-manifest.json', raw)
|
|
state = {'schema': 1, 'id': tx.name, 'release': identity, 'version': document['version'],
|
|
'old_version': old_version, 'new_version': document['version'],
|
|
'was_active': True,
|
|
'runtime_requested': document.get('application_update') == RUNTIME_POLICY,
|
|
'new': pair_identity(tx / 'new-binary', tx / 'new-frontend')}
|
|
return schedule_transaction(tx, state, bundle)
|
|
|
|
|
|
def schedule_transaction(tx, state, bundle):
|
|
# Frozen authenticated inputs and the installed owner's source; the caller
|
|
# holds the root lock. No candidate-supplied worker is executed.
|
|
runner = TRANSACTIONS / ('worker-' + hashlib.sha256(bundle.encode()).hexdigest() + '.py')
|
|
if runner.exists():
|
|
if secure_read(runner, 1024**2) != bundle.encode():
|
|
raise ValueError('worker code identity changed')
|
|
else:
|
|
write_manifest(runner, bundle.encode())
|
|
_, previous = current()
|
|
check_owner_dropins(previous)
|
|
recovery_unit = 'archipelago-public-recovery-' + tx.name + '.service'
|
|
worker_unit = 'archipelago-public-worker-' + tx.name + '.service'
|
|
dropin = f'[Unit]\nRequires={recovery_unit}\nAfter={recovery_unit}\n'
|
|
state.update(dropin=dropin, worker_unit=worker_unit,
|
|
runtime_startup_guard=bool(state.get('runtime_requested')),
|
|
previous_id=previous['id'] if previous else None)
|
|
state = journal(tx, state, 'queued')
|
|
sync_directory(TRANSACTIONS)
|
|
os.close(trusted_open(UNIT_DIR, directory=True))
|
|
mounts = 'RequiresMountsFor=/var/lib/archipelago /var/lib/archipelago-public-transactions /opt/archipelago\n'
|
|
before = ' '.join(('basic.target', 'archipelago.service', *MAINTENANCE_UNITS))
|
|
recovery_prefix = ('[Unit]\nDescription=Recover interrupted public application transaction\n'
|
|
'DefaultDependencies=no\nAfter=local-fs.target\n' + mounts + f'Before={before}\n'
|
|
'[Service]\nType=oneshot\nRemainAfterExit=yes\nTimeoutStartSec=900\n')
|
|
# Validate/arm this boot's dependency without replaying an upgrade whose
|
|
# pointer is not published yet. Its persistent boot command is installed
|
|
# and enabled BEFORE the pointer can make the transaction authoritative.
|
|
# The live arm must not wait behind the manager's in-progress start job.
|
|
# Its persistent boot ordering is installed durably before current below.
|
|
arm_prefix = recovery_prefix.replace(f'Before={before}\n', '') if state.get('adoption') else recovery_prefix
|
|
atomic(UNIT_DIR / recovery_unit, (arm_prefix +
|
|
f'ExecStart=/usr/bin/python3 -I {runner} arm {tx.name}\n').encode(), 0o644)
|
|
atomic(UNIT_DIR / worker_unit, (
|
|
'[Unit]\nDescription=Independent public application transaction worker\n' + mounts +
|
|
'StartLimitIntervalSec=600\nStartLimitBurst=3\n[Service]\nType=exec\n'
|
|
'ExecStartPre=/usr/bin/sleep 2\n'
|
|
f'ExecStart=/usr/bin/python3 -I {runner} run {tx.name}\nRestart=on-failure\nRestartSec=2\n'
|
|
'TimeoutStopSec=10\nRuntimeMaxSec=900\nUMask=0077\n').encode(), 0o644)
|
|
# Fresh transaction unit names are loaded on demand by start. During
|
|
# adoption, a daemon-reload would reset the starting manager's timeout.
|
|
if not state.get('adoption'):
|
|
command('/usr/bin/systemctl', 'daemon-reload')
|
|
command('/usr/bin/systemctl', 'start', recovery_unit)
|
|
atomic(UNIT_DIR / recovery_unit, (recovery_prefix +
|
|
f'ExecStart=/usr/bin/python3 -I {runner} recover-boot {tx.name}\n').encode(), 0o644)
|
|
retain_boot_recovery(recovery_unit)
|
|
if not state.get('adoption'):
|
|
command('/usr/bin/systemctl', 'daemon-reload')
|
|
atomic(TRANSACTIONS / 'current', tx.name.encode())
|
|
install_owner_dropins(state, reload=not state.get('adoption'))
|
|
command('/usr/bin/systemctl', 'start', '--no-block', worker_unit)
|
|
return state
|
|
|
|
|
|
def worker_main():
|
|
if os.geteuid() != 0:
|
|
raise PermissionError('root public pair worker required')
|
|
os.umask(0o077)
|
|
if sys.argv[1:] == ['apply']:
|
|
# Source-only bootstrap for a legacy binary with ceremony verification
|
|
# but without the new backend RPC dispatch. The reviewed bundle must be
|
|
# installed under root custody first; never run it from a downloaded UI.
|
|
import io
|
|
bundle = globals().get('BUNDLE')
|
|
if bundle is None:
|
|
bundle = secure_read(Path(__file__), 1024**2).decode()
|
|
sys.argv = [sys.argv[0]]
|
|
output = io.StringIO()
|
|
with contextlib.redirect_stdout(output):
|
|
main() # fixed-root preparation, original signed JSON on stdin
|
|
prepared = json.loads(output.getvalue())
|
|
print(json.dumps(submit(Path(prepared['path']).name, bundle)))
|
|
elif len(sys.argv) == 3 and sys.argv[1] == 'submit':
|
|
bundle = globals().get('BUNDLE')
|
|
if bundle is None:
|
|
bundle = secure_read(Path(__file__), 1024**2).decode()
|
|
print(json.dumps(submit(sys.argv[2], bundle)))
|
|
elif sys.argv[1:] == ['adopt']:
|
|
bundle = globals().get('BUNDLE')
|
|
if bundle is None:
|
|
bundle = secure_read(Path(__file__), 1024**2).decode()
|
|
print(json.dumps(adopt_legacy_pair(bundle)))
|
|
elif sys.argv[1:] == ['run'] or (len(sys.argv) == 3 and sys.argv[1] == 'run'
|
|
and re.fullmatch('[0-9a-f]{32}', sys.argv[2])):
|
|
run_worker(sys.argv[2] if len(sys.argv) == 3 else None)
|
|
elif len(sys.argv) == 3 and sys.argv[1] == 'arm':
|
|
tx, state = load_transaction(sys.argv[2])
|
|
require_legacy_service()
|
|
if state['phase'] != 'queued' or pair_identity(tx / 'new-binary', tx / 'new-frontend') != state['new']:
|
|
raise ValueError('unprepared public recovery dependency')
|
|
if state.get('adoption'):
|
|
check_preimages(tx, state)
|
|
elif sys.argv[1:] == ['request-recovery']:
|
|
with owner_lock():
|
|
require_legacy_service()
|
|
tx, state = current()
|
|
if not state or state['phase'] in ('restored', 'aborted'):
|
|
raise ValueError('no public pair available for recovery')
|
|
state['recovery_requested'] = True
|
|
journal(tx, state, 'recovery-requested' if state['phase'] == 'committed' else state['phase'])
|
|
command('/usr/bin/systemctl', 'start', '--no-block', state['worker_unit'])
|
|
elif len(sys.argv) == 3 and sys.argv[1] == 'recover-boot':
|
|
own_tx, own = load_transaction(sys.argv[2])
|
|
_, selected = current()
|
|
if selected and selected['id'] != own['id']:
|
|
# An old enabled recovery unit must never operate on a later job.
|
|
# Only a durable, not-yet-published admission whose predecessor is
|
|
# still current may acquire its pointer after a scheduling crash.
|
|
if (own['phase'] != 'queued' or own.get('previous_id') != selected['id']
|
|
or selected['phase'] not in TERMINAL):
|
|
return
|
|
elif selected is None and own.get('previous_id') is not None:
|
|
raise ValueError('public recovery predecessor disappeared')
|
|
with owner_lock():
|
|
_, selected = current()
|
|
if not selected or selected['id'] != own['id']:
|
|
if (own['phase'] != 'queued' or
|
|
(selected and (selected['id'] != own.get('previous_id') or selected['phase'] not in TERMINAL))):
|
|
raise ValueError('public recovery pointer changed')
|
|
atomic(TRANSACTIONS / 'current', own['id'].encode())
|
|
tx, state = current()
|
|
if state['phase'] not in TERMINAL:
|
|
recovery(tx, state, True)
|
|
elif sys.argv[1:] == ['status']:
|
|
_, state = current()
|
|
print(json.dumps(state))
|
|
else:
|
|
raise ValueError('fixed public worker command required')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
worker_main()
|