225 lines
9.4 KiB
Python
225 lines
9.4 KiB
Python
#!/usr/bin/python3 -I
|
|
"""Prepare an inert, root-custody OS-owner generation using independent admin pins.
|
|
|
|
No service activation or transaction authority is issued here. The OS owner must
|
|
consume/revalidate this generation through its separately qualified contract.
|
|
No network, subprocess, package-manager, data restore or legacy updater fallback.
|
|
"""
|
|
import argparse
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
|
|
BASE = 'var/lib/archipelago-os-bootstrap'
|
|
BLOCKERS = ('var/lib/archipelago-d09/ledger.json',
|
|
'etc/archipelago-d09/binding.json',
|
|
'var/lib/archipelago-os-update/ledger.json')
|
|
OS_TRANSACTIONS = 'var/lib/archipelago/os-update/transactions'
|
|
OS_TRANSACTION_ROOTS = (OS_TRANSACTIONS, 'var/lib/archipelago-os-update/transactions')
|
|
|
|
|
|
def digest(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def decode(raw):
|
|
def pairs(items):
|
|
result = {}
|
|
for key, value in items:
|
|
if key in result:
|
|
raise ValueError('duplicate JSON field')
|
|
result[key] = value
|
|
return result
|
|
return json.loads(raw, object_pairs_hook=pairs)
|
|
|
|
|
|
def pin(value):
|
|
if not isinstance(value, str) or not re.fullmatch('[0-9a-f]{64}', value):
|
|
raise ValueError('SHA256 pin required')
|
|
return value
|
|
|
|
|
|
def relative(value):
|
|
if not isinstance(value, str) or not value or any(
|
|
p in ('', '.', '..') for p in value.split('/')) or value.startswith('/'):
|
|
raise ValueError('canonical relative path required')
|
|
return value
|
|
|
|
|
|
def secure(path, uid=0, directory=False):
|
|
"""Every ancestor must be trusted; do not resolve attacker-controlled links."""
|
|
path = Path(path).absolute()
|
|
for part in reversed((path, *path.parents)):
|
|
s = part.lstat()
|
|
if stat.S_ISLNK(s.st_mode) or s.st_uid != uid or s.st_mode & 0o022:
|
|
raise ValueError('unprotected custody: ' + str(part))
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
s = os.fstat(fd)
|
|
if (not (stat.S_ISDIR(s.st_mode) if directory else stat.S_ISREG(s.st_mode))
|
|
or (not directory and s.st_nlink != 1)):
|
|
os.close(fd)
|
|
raise ValueError('regular unaliased object required')
|
|
return fd
|
|
|
|
|
|
def read(path):
|
|
fd = secure(path)
|
|
with os.fdopen(fd, 'rb') as stream:
|
|
return stream.read()
|
|
|
|
|
|
def sync(path):
|
|
fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def write(path, raw):
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
with os.fdopen(fd, 'wb') as stream:
|
|
stream.write(raw)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
|
|
|
|
def version(value):
|
|
# Intentionally narrow transition grammar, not a general SemVer comparator.
|
|
match = re.fullmatch(r'(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)\.(0|[1-9][0-9]*)-alpha(?:\+uat)?', value)
|
|
if not match:
|
|
raise ValueError('unsupported version channel')
|
|
return tuple(map(int, match.groups()))
|
|
|
|
|
|
def validate(plan):
|
|
if set(plan) != {'schema', 'machine_id_sha256', 'current', 'candidate',
|
|
'admin', 'observations', 'files'} or plan['schema'] != 1:
|
|
raise ValueError('bootstrap schema')
|
|
pin(plan['machine_id_sha256'])
|
|
for name in ('current', 'candidate'):
|
|
item = plan[name]
|
|
if set(item) != {'version', 'source_sha256', 'backend_sha256'}:
|
|
raise ValueError('release identity fields')
|
|
version(item['version'])
|
|
pin(item['source_sha256']); pin(item['backend_sha256'])
|
|
if version(plan['candidate']['version']) <= version(plan['current']['version']):
|
|
raise ValueError('downgrade or same-version replacement refused')
|
|
admin = plan['admin']
|
|
if set(admin) != {'uid', 'service_uid', 'access_receipt_sha256'} or any(
|
|
type(admin[k]) is not int or admin[k] <= 0 for k in ('uid', 'service_uid')):
|
|
raise ValueError('independent administrator required')
|
|
if admin['uid'] == admin['service_uid']:
|
|
raise ValueError('shared service administrator refused')
|
|
pin(admin['access_receipt_sha256'])
|
|
# Observation paths are read-only, independently pinned evidence. They do
|
|
# not confer service-drain, runtime exclusion or financial restore authority.
|
|
if not isinstance(plan['observations'], dict) or not plan['observations']:
|
|
raise ValueError('installed inventory pins required')
|
|
for path, sha in plan['observations'].items():
|
|
relative(path); pin(sha)
|
|
if admin['access_receipt_sha256'] not in plan['observations'].values():
|
|
raise ValueError('administrator access receipt must be an observed pinned input')
|
|
files = plan['files']
|
|
if not isinstance(files, dict) or not files or len(files) > 256:
|
|
raise ValueError('bounded owner files required')
|
|
for name, sha in files.items():
|
|
if not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}', name):
|
|
raise ValueError('flat owner payload names required')
|
|
if name in ('control.json', 'approval.json'):
|
|
raise ValueError('reserved control name')
|
|
pin(sha)
|
|
source = json.dumps(files, sort_keys=True, separators=(',', ':')).encode()
|
|
if digest(source) != plan['candidate']['source_sha256']:
|
|
raise ValueError('candidate source inventory digest differs')
|
|
|
|
|
|
def prepare(root, approval, expected, payload):
|
|
if os.geteuid() != 0:
|
|
raise ValueError('independent root administrator required')
|
|
root = Path(root).absolute()
|
|
fd = secure(root, directory=True); os.close(fd)
|
|
raw = read(approval)
|
|
if digest(raw) != pin(expected):
|
|
raise ValueError('independently authenticated approval digest differs')
|
|
plan = decode(raw)
|
|
validate(plan)
|
|
if digest(read(root / 'etc/machine-id')) != plan['machine_id_sha256']:
|
|
raise ValueError('different installed machine')
|
|
if digest(read(root / 'usr/local/bin/archipelago')) != plan['current']['backend_sha256']:
|
|
raise ValueError('installed backend drift')
|
|
# Version/source provenance is an independently authenticated admin input,
|
|
# bound to exact executable bytes, never trusted from updater state.json.
|
|
for path, sha in plan['observations'].items():
|
|
if digest(read(root / path)) != sha:
|
|
raise ValueError('installed observation drift: ' + path)
|
|
data = {}
|
|
for name, sha in plan['files'].items():
|
|
data[name] = read(Path(payload) / name)
|
|
if digest(data[name]) != sha:
|
|
raise ValueError('owner source hash drift: ' + name)
|
|
base = root / BASE
|
|
fd = secure(base.parent, directory=True); os.close(fd)
|
|
base.mkdir(mode=0o700, exist_ok=True)
|
|
fd = secure(base, directory=True); os.close(fd)
|
|
lock = os.open(base / 'lock', os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
fd = secure(base / 'lock'); os.close(fd)
|
|
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
for name in BLOCKERS:
|
|
path = root / name
|
|
# A symlink/missing inaccessible ancestor is not an empty ledger.
|
|
for parent in reversed(path.parents):
|
|
if parent == root or root in parent.parents:
|
|
if parent.exists() or parent.is_symlink():
|
|
fd = secure(parent, directory=True); os.close(fd)
|
|
if path.exists() or path.is_symlink():
|
|
raise ValueError('existing authority requires external reconciliation: ' + name)
|
|
for name in OS_TRANSACTION_ROOTS:
|
|
transactions = root / name
|
|
if transactions.exists() or transactions.is_symlink():
|
|
fd = secure(transactions, directory=True); os.close(fd)
|
|
if any(transactions.iterdir()):
|
|
raise ValueError('existing OS transactions require external reconciliation')
|
|
if (base / 'prepared').exists() or (base / 'prepared').is_symlink():
|
|
raise ValueError('existing bootstrap; no automatic replacement/recovery')
|
|
stage = base / ('pending-' + expected)
|
|
# Never resume/erase an interrupted generation automatically.
|
|
if list(base.glob('pending-*')):
|
|
raise ValueError('unresolved bootstrap generation')
|
|
stage.mkdir(mode=0o700)
|
|
sync(base)
|
|
write(stage / 'approval.json', raw)
|
|
for name, content in sorted(data.items()):
|
|
write(stage / name, content)
|
|
control = {'schema': 1, 'state': 'PREPARED_NOT_ACTIVATED',
|
|
'approval_sha256': expected, 'current': plan['current'],
|
|
'candidate': plan['candidate'], 'files': plan['files'],
|
|
'rollback_scope': 'code-only; no financial or persistent data'}
|
|
write(stage / 'control.json', json.dumps(control, sort_keys=True).encode())
|
|
sync(stage)
|
|
os.rename(stage, base / 'prepared')
|
|
sync(base)
|
|
return control
|
|
finally:
|
|
os.close(lock)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--root', type=Path, default=Path('/'))
|
|
parser.add_argument('--approval', type=Path, required=True)
|
|
parser.add_argument('--approval-sha256', required=True,
|
|
help='pin received over the independent administrator channel')
|
|
parser.add_argument('--payload', type=Path, required=True)
|
|
args = parser.parse_args()
|
|
print(json.dumps(prepare(args.root, args.approval, args.approval_sha256, args.payload), sort_keys=True))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|