244 lines
12 KiB
Python
Executable File
244 lines
12 KiB
Python
Executable File
#!/usr/bin/python3 -I
|
|
"""Pinned Unix client. Public owner operations are only submit and inspect."""
|
|
import argparse
|
|
import fcntl
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import secrets
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
|
|
from d09_wire import MAX, bind_inspect, bind_submit, canonical, decode, digest, release_digest
|
|
from d09_store import read
|
|
|
|
SOCKET = '/run/archipelago-d09/lifecycle.sock'
|
|
PUBLIC = pathlib.Path('/etc/archipelago-d09/client.json')
|
|
CACHE = pathlib.Path('/var/lib/archipelago/d09-requests')
|
|
|
|
|
|
def exchange(request, record):
|
|
# Root-only socket ancestry plus SO_PEERCRED pins this live exchange to the
|
|
# protected owner. JSON PID/hash fields are never authentication.
|
|
for path in (pathlib.Path('/run'), pathlib.Path('/run/archipelago-d09')):
|
|
st = path.lstat()
|
|
if not stat.S_ISDIR(st.st_mode) or st.st_uid != 0 or st.st_mode & 0o022:
|
|
raise PermissionError('unsafe owner socket ancestry')
|
|
st = os.lstat(SOCKET)
|
|
if not stat.S_ISSOCK(st.st_mode) or st.st_uid != 0:
|
|
raise PermissionError('unsafe owner socket')
|
|
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
|
|
conn.settimeout(4)
|
|
conn.connect(SOCKET)
|
|
pid, uid, gid = struct.unpack('3i', conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12))
|
|
if uid != 0:
|
|
raise PermissionError('unauthenticated owner')
|
|
conn.sendall(canonical(request))
|
|
conn.shutdown(socket.SHUT_WR)
|
|
raw = b''
|
|
while True:
|
|
part = conn.recv(8192)
|
|
if not part:
|
|
break
|
|
raw += part
|
|
if len(raw) > MAX:
|
|
raise ValueError('oversized response')
|
|
result = decode(record, raw)
|
|
if record == 'inspect_response':
|
|
return bind_inspect(request, result)
|
|
return bind_submit(request, result)
|
|
|
|
|
|
def inspect(binding):
|
|
return exchange(dict(schema_version=1, target_binding_digest=binding['target_binding_digest'],
|
|
transaction_id=None, challenge=secrets.token_hex(32)), 'inspect_response')
|
|
|
|
|
|
def projection(binding, response):
|
|
r = response['receipt']
|
|
if r and any(r[k] != binding[k] for k in ('transaction_id', 'target_binding_digest', 'composition_digest')):
|
|
raise PermissionError('client binding differs from authenticated disposition')
|
|
state = 'UNKNOWN'
|
|
if response.get('status') == 'NO_D09_TRANSACTION':
|
|
state = 'NO_D09_TRANSACTION'
|
|
elif r:
|
|
state = 'NARROW_ACTIVATION_OBSERVED' if r['outcome'] == 'COMPLETE' else r['phase'] if r['phase'] in ('CONFLICT', 'READMISSION_REQUIRED') else 'PENDING'
|
|
restored = r is not None and r['config_action'] == 'RESTORE'
|
|
artifacts = binding['restore_artifacts' if restored else 'artifacts']
|
|
return dict(schema_version=1, state=state, reason=response['reason'],
|
|
transaction_id=r['transaction_id'] if r else None,
|
|
target_binding_digest=binding['target_binding_digest'], composition_digest=binding['composition_digest'],
|
|
receipt_digest=response['receipt_digest'], sequence=r['sequence'] if r else 0,
|
|
service_invocation_id=r['observed_invocation_id'] if r else None,
|
|
observed_at=time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()) if r and r['outcome'] == 'COMPLETE' else None,
|
|
receiving_backend_sha256=artifacts['backend'], paired_frontend_sha256=artifacts['frontend'],
|
|
selected_release_digest=release_digest(binding['release_identity']),
|
|
config_action=r['config_action'] if r else 'NONE',
|
|
staging_generation=r['staging_generation'] if r else None,
|
|
os_update=response.get('os_update'))
|
|
|
|
|
|
def cache_read(path):
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
with os.fdopen(fd, 'rb') as source:
|
|
st = os.fstat(source.fileno())
|
|
if not stat.S_ISREG(st.st_mode) or st.st_uid != os.geteuid() or st.st_mode & 0o077:
|
|
raise PermissionError('unsafe retry record')
|
|
raw = source.read(MAX + 1)
|
|
if len(raw) > MAX:
|
|
raise ValueError('oversized retry record')
|
|
return raw
|
|
|
|
|
|
def cache_write(path, raw, replace=False):
|
|
# Publish whole fsynced bytes. Immutable request/disposition records survive
|
|
# active-pointer rotation; a crash cannot leave a truncated active request.
|
|
fd, name = tempfile.mkstemp(prefix='.retry-', dir=CACHE)
|
|
with os.fdopen(fd, 'wb') as dest:
|
|
os.fchmod(dest.fileno(), 0o600)
|
|
dest.write(raw)
|
|
dest.flush()
|
|
os.fsync(dest.fileno())
|
|
if replace:
|
|
os.replace(name, path)
|
|
else:
|
|
try:
|
|
os.link(name, path, follow_symlinks=False)
|
|
except FileExistsError:
|
|
if cache_read(path) != raw:
|
|
raise ValueError('immutable retry evidence differs')
|
|
os.unlink(name)
|
|
dfd = os.open(CACHE, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
os.fsync(dfd)
|
|
finally:
|
|
os.close(dfd)
|
|
|
|
|
|
def convenience(binding, operation, reason, selected_release=None, staging_generation=None):
|
|
if operation == 'prepare':
|
|
# Before inspect, cache publication, reservation or any grant consumption.
|
|
if release_digest(selected_release) != release_digest(binding['release_identity']):
|
|
raise PermissionError('selected release differs from external admission')
|
|
if type(staging_generation) is not str or not re.fullmatch('[0-9a-f]{32}', staging_generation) or int(staging_generation, 16) == 0:
|
|
raise ValueError('exact staging generation required')
|
|
route = 'UPDATER' if operation == 'prepare' else {'automatic': 'AUTOMATIC', 'manual': 'MANUAL', 'guard': 'GUARD'}[reason]
|
|
invocation = os.environ.get('INVOCATION_ID')
|
|
# Durable retry bytes belong to the caller, not the authority. Their exact
|
|
# request/acceptance is checked by the owner; loss never permits reset there.
|
|
CACHE.mkdir(mode=0o700, exist_ok=True)
|
|
st = CACHE.lstat()
|
|
if not stat.S_ISDIR(st.st_mode) or st.st_uid != os.geteuid() or st.st_mode & 0o077:
|
|
raise PermissionError('unsafe retry cache')
|
|
token = digest(canonical([binding['transaction_id'], route, invocation]))
|
|
path = CACHE / (token + '.json')
|
|
lock = os.open(CACHE / (token + '.lock'), os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
|
|
try:
|
|
st = os.fstat(lock)
|
|
if not stat.S_ISREG(st.st_mode) or st.st_uid != os.geteuid() or st.st_mode & 0o077:
|
|
raise PermissionError('unsafe retry lock')
|
|
# Serializes inspect/submit/disposition/rotation among convenience callers.
|
|
fcntl.flock(lock, fcntl.LOCK_EX)
|
|
return submit_cached(binding, operation, route, invocation, token, path, staging_generation)
|
|
finally:
|
|
os.close(lock)
|
|
|
|
|
|
def submit_cached(binding, operation, route, invocation, token, path, staging_generation=None):
|
|
fresh = False
|
|
previous_id = None
|
|
# One bounded fresh retry resolves the normal inspect/worker-advance race.
|
|
# A second refusal remains retryable on the next invocation of this client.
|
|
for attempt in range(2):
|
|
status = inspect(binding)
|
|
if operation == 'prepare':
|
|
plans = [c for c in binding['release_identity'][1] if c[0] == 'os-update-plan.json']
|
|
if plans:
|
|
os_status = status.get('os_update')
|
|
if not os_status:
|
|
raise PermissionError('OS updater upgrade required; provision the protected OS updater')
|
|
if os_status['phase'] in ('UNKNOWN', 'RECONCILIATION_REQUIRED', 'APPLY_INTENT', 'VERIFY'):
|
|
raise PermissionError('OS transaction requires administrator reconciliation; do not repeat dpkg')
|
|
if not os_status['supported']:
|
|
raise PermissionError('OS updater upgrade required; provision the protected OS updater')
|
|
if os_status['plan_sha256'] != plans[0][2]:
|
|
raise PermissionError('OS plan not admitted; admit the exact signed release and maintenance operation')
|
|
if status['status'] == 'UNKNOWN':
|
|
raise PermissionError('continuity unavailable')
|
|
try:
|
|
raw = cache_read(path)
|
|
except FileNotFoundError:
|
|
raw = None
|
|
if raw is not None and not fresh:
|
|
request = decode('submit_request', raw)
|
|
if operation == 'prepare' and request['staging_generation'] != staging_generation:
|
|
raise PermissionError('retained request belongs to another staging generation')
|
|
else:
|
|
request_id = secrets.token_hex(16)
|
|
if request_id == previous_id:
|
|
raise ValueError('fresh request identity collision')
|
|
request = dict(schema_version=1, operation='APPLY' if operation == 'prepare' else 'RECOVER', route=route,
|
|
transaction_id=binding['transaction_id'], request_id=request_id,
|
|
target_binding_digest=binding['target_binding_digest'], composition_digest=binding['composition_digest'],
|
|
grant_digest=binding['grants'][route], expected_sequence=status['head_sequence'],
|
|
requester_invocation_id=invocation, observer_receipt_digest=None,
|
|
selected_release_digest=release_digest(binding['release_identity']) if operation == 'prepare' else None,
|
|
staging_generation=staging_generation if operation == 'prepare' else None,
|
|
reason={'UPDATER': 'APPLY_REQUEST', 'AUTOMATIC': 'AUTOMATIC_FAILURE', 'MANUAL': 'MANUAL_REQUEST', 'GUARD': 'CRASH_NOTICE'}[route])
|
|
raw = canonical(request)
|
|
decode('submit_request', raw)
|
|
stem = token + '.' + request['request_id']
|
|
cache_write(CACHE / (stem + '.request.json'), raw)
|
|
if fresh or not path.exists():
|
|
cache_write(path, raw, replace=True)
|
|
# Transport loss, EOF, malformed responses and UNKNOWN never rotate.
|
|
result = bind_submit(request, exchange(request, 'submit_response'))
|
|
accepted = CACHE / (stem + '.acceptance.json')
|
|
if result['status'] == 'ACCEPTED':
|
|
cache_write(accepted, canonical(result['acceptance']))
|
|
return result
|
|
if result['request_digest'] == digest(raw):
|
|
evidence = canonical({'request': request, 'response': result})
|
|
cache_write(CACHE / (stem + '.refusal-' + digest(evidence) + '.json'), evidence)
|
|
# These reasons occur only AFTER authenticated continuity and per-key
|
|
# lookup. They prove this exact key never accepted, even when RETAINED
|
|
# refers to another request's active reservation. Other refusals cannot
|
|
# authorize replacement. In particular UNKNOWN/IDEMPOTENCY_MISMATCH and
|
|
# any locally known acceptance keep their original exact retry bytes.
|
|
definitive = (result['request_digest'] == digest(raw)
|
|
and result['reservation'] in ('NONE', 'RETAINED')
|
|
and result['reason'] in ('STALE_SEQUENCE', 'GRANT_INVALID')
|
|
and not accepted.exists())
|
|
if not definitive or attempt == 1:
|
|
raise PermissionError(result['reason'])
|
|
previous_id = request['request_id']
|
|
fresh = True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('operation', choices=('inspect', 'submit', 'prepare', 'recover'))
|
|
parser.add_argument('--reason', choices=('automatic', 'manual', 'guard'))
|
|
parser.add_argument('--selected-release')
|
|
parser.add_argument('--staging-generation')
|
|
args = parser.parse_args()
|
|
binding = json.loads(read(PUBLIC))
|
|
if args.operation == 'submit':
|
|
request = decode('submit_request', sys.stdin.buffer.read(MAX + 1))
|
|
result = exchange(request, 'submit_response')
|
|
else:
|
|
selected = json.loads(args.selected_release) if args.selected_release else None
|
|
response = inspect(binding) if args.operation == 'inspect' else convenience(binding, args.operation, args.reason, selected, args.staging_generation)
|
|
result = projection(binding, response)
|
|
sys.stdout.buffer.write(canonical(result))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|