302 lines
14 KiB
Python
302 lines
14 KiB
Python
"""Protected durable D09 ledger, with a live-owner continuity anchor.
|
|
|
|
No missing-store initialization, disk-seeded restart anchor, or public reset.
|
|
All callers serialize access; the outside owner keeps this object alive.
|
|
"""
|
|
import copy
|
|
import os
|
|
import pathlib
|
|
import stat
|
|
import tempfile
|
|
|
|
from d09_wire import canonical, digest, validate
|
|
|
|
|
|
def secure(path, directory=False):
|
|
path = pathlib.Path(path)
|
|
for parent in reversed(path.parents):
|
|
st = parent.lstat()
|
|
if not stat.S_ISDIR(st.st_mode) or st.st_uid != 0 or st.st_mode & 0o022:
|
|
raise PermissionError('unprotected ancestry')
|
|
flags = os.O_RDONLY | os.O_NOFOLLOW | (os.O_DIRECTORY if directory else 0)
|
|
fd = os.open(path, flags)
|
|
st = os.fstat(fd)
|
|
if st.st_uid != 0 or st.st_mode & 0o022 or (not directory and (not stat.S_ISREG(st.st_mode) or st.st_nlink != 1)):
|
|
os.close(fd)
|
|
raise PermissionError('unprotected object')
|
|
return fd
|
|
|
|
|
|
def read(path):
|
|
fd = secure(path)
|
|
with os.fdopen(fd, 'rb') as source:
|
|
return source.read()
|
|
|
|
|
|
def identity(fd):
|
|
st = os.fstat(fd)
|
|
info = pathlib.Path('/proc/self/fdinfo/%d' % fd).read_text()
|
|
mount = next(line.split()[1] for line in info.splitlines() if line.startswith('mnt_id:'))
|
|
return (st.st_dev, st.st_ino, st.st_uid, st.st_gid, st.st_mode, st.st_nlink, mount)
|
|
|
|
|
|
def sync_dir(path):
|
|
fd = secure(path, True)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def atomic(path, raw, mode=0o600):
|
|
path = pathlib.Path(path)
|
|
parent = secure(path.parent, True)
|
|
name = '.d09-' + os.urandom(16).hex()
|
|
fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, mode, dir_fd=parent)
|
|
try:
|
|
with os.fdopen(fd, 'wb') as dest:
|
|
dest.write(raw)
|
|
dest.flush()
|
|
os.fsync(dest.fileno())
|
|
os.rename(name, path.name, src_dir_fd=parent, dst_dir_fd=parent)
|
|
os.fsync(parent)
|
|
finally:
|
|
# An uncertain publication is evidence, never silently removed.
|
|
os.close(parent)
|
|
|
|
|
|
def initial(binding):
|
|
return {'sequence': 0, 'receipt': None, 'requests': {}, 'inbox': [],
|
|
'consumed': [], 'epochs': [binding['admission_epoch_id']],
|
|
'grants': [], 'epoch': binding['admission_epoch_id'],
|
|
'transaction_id': binding['transaction_id'], 'intent': None,
|
|
'execution': None, 'readmissions': [], 'journal': [], 'intents': [],
|
|
'notice_grant': None, 'execution_grant': None, 'prerequisite': None,
|
|
'os_admission': None, 'os_observations': [], 'os_checkpoints': [],
|
|
'manager_events': [], 'manager_fault': None}
|
|
|
|
|
|
class ContinuityError(Exception):
|
|
pass
|
|
|
|
|
|
class Store:
|
|
def __init__(self, root, binding, admitted=False):
|
|
self.root, self.binding = pathlib.Path(root), binding
|
|
self.fd = secure(self.root, True)
|
|
self.identity = identity(self.fd)
|
|
self.head_fd = secure(self.root / 'ledger.json')
|
|
self.head_identity = identity(self.head_fd)
|
|
self.raw = read(self.root / 'ledger.json')
|
|
import json
|
|
self.value = json.loads(self.raw)
|
|
self.binding = self.value.get('active_binding', binding)
|
|
self.anchor = digest(self.raw) if admitted else None
|
|
self.check_chain()
|
|
|
|
def check_chain(self):
|
|
v = self.value
|
|
if v['transaction_id'] != self.binding['transaction_id']:
|
|
raise ContinuityError('target index')
|
|
previous = None
|
|
previous_intent = None
|
|
for seq, hashed in enumerate(v['journal'], 1):
|
|
import json
|
|
raw = read(self.root / ('receipt-%016d.json' % seq))
|
|
r = json.loads(raw)
|
|
validate('receipt', r)
|
|
if digest(raw) != hashed or r['sequence'] != seq or r['previous_receipt_digest'] != previous:
|
|
raise ContinuityError('journal chain')
|
|
intent_raw = read(self.root / ('intent-%016d.json' % seq))
|
|
intent = json.loads(intent_raw)
|
|
if digest(intent_raw) != v['intents'][seq - 1] or intent['sequence'] != seq or intent['receipt_digest'] != hashed or intent['previous_intent_digest'] != previous_intent:
|
|
raise ContinuityError('intent chain')
|
|
previous_intent = digest(intent_raw)
|
|
previous = hashed
|
|
if v['sequence'] != len(v['journal']) or len(v['intents']) != len(v['journal']) or (v['receipt'] and digest(canonical(v['receipt'])) != previous):
|
|
raise ContinuityError('head')
|
|
for entry in v['requests'].values():
|
|
req, a = entry['request'], entry['acceptance']
|
|
validate('submit_request', req)
|
|
validate('acceptance', a)
|
|
if digest(canonical(req)) != a['request_digest'] or a['accepted_sequence'] > v['sequence']:
|
|
raise ContinuityError('acceptance')
|
|
|
|
def check(self):
|
|
for path, expected, directory in ((self.root, self.identity, True),
|
|
(self.root / 'ledger.json', self.head_identity, False)):
|
|
fd = secure(path, directory)
|
|
try:
|
|
if identity(fd) != expected:
|
|
self.anchor = None
|
|
finally:
|
|
os.close(fd)
|
|
if self.anchor is None or digest(read(self.root / 'ledger.json')) != self.anchor:
|
|
self.anchor = None
|
|
raise ContinuityError('owner continuity lost')
|
|
self.check_chain()
|
|
|
|
def commit(self, value, receipt):
|
|
self.check()
|
|
value = copy.deepcopy(value)
|
|
receipt = copy.deepcopy(receipt)
|
|
receipt['sequence'] = self.value['sequence'] + 1
|
|
if receipt['sequence'] > 9007199254740991:
|
|
raise OverflowError('SEQUENCE_EXHAUSTED')
|
|
receipt['previous_receipt_digest'] = self.value['journal'][-1] if self.value['journal'] else None
|
|
validate('receipt', receipt)
|
|
path = self.root / ('receipt-%016d.json' % receipt['sequence'])
|
|
intent_path = self.root / ('intent-%016d.json' % receipt['sequence'])
|
|
if path.exists() or intent_path.exists():
|
|
# Crash after record publication before head is an uncommitted tail.
|
|
# Preserve it and require external continuity resolution.
|
|
self.anchor = None
|
|
raise ContinuityError('uncommitted tail')
|
|
raw = canonical(receipt)
|
|
intent = canonical(dict(sequence=receipt['sequence'], receipt_digest=digest(raw),
|
|
previous_intent_digest=self.value['intents'][-1] if self.value['intents'] else None,
|
|
transaction_id=value['transaction_id'], intent=value['intent'],
|
|
execution=value['execution'], consumed=value['consumed']))
|
|
atomic(intent_path, intent)
|
|
atomic(path, raw)
|
|
value['journal'].append(digest(raw))
|
|
value['intents'].append(digest(intent))
|
|
value['sequence'], value['receipt'] = receipt['sequence'], receipt
|
|
raw = canonical(value)
|
|
try:
|
|
atomic(self.root / 'ledger.json', raw)
|
|
except BaseException:
|
|
self.anchor = None
|
|
raise
|
|
self.value, self.raw, self.anchor = value, raw, digest(raw)
|
|
os.close(self.head_fd)
|
|
self.head_fd = secure(self.root / 'ledger.json')
|
|
self.head_identity = identity(self.head_fd)
|
|
return receipt
|
|
|
|
def commit_prerequisite(self, record):
|
|
"""Grant-A consumption before any D09 reservation; never invent a receipt.
|
|
|
|
This lives in the same protected, live-anchored head as the later grant-B
|
|
transaction. Owner loss still requires external resolution, even at zero.
|
|
"""
|
|
self.check()
|
|
value = copy.deepcopy(self.value)
|
|
if value['sequence'] != 0 or value['receipt'] is not None or any(
|
|
value[k] for k in ('requests', 'inbox', 'journal', 'intents', 'readmissions')):
|
|
raise ContinuityError('prerequisite activation requires an empty target')
|
|
previous = value.get('prerequisite')
|
|
if previous is None:
|
|
if record['state'] != 'START_INTENT' or record['grant_digest'] in value['grants']:
|
|
raise ValueError('prerequisite grant already consumed')
|
|
value['grants'].append(record['grant_digest'])
|
|
elif previous['state'] != 'START_INTENT' or record['state'] not in ('STARTED', 'START_UNCERTAIN') or any(
|
|
previous[k] != record[k] for k in ('grant_digest', 'owner_incarnation', 'boot_id', 'backend_sha256')):
|
|
raise ValueError('invalid prerequisite transition')
|
|
value['prerequisite'] = copy.deepcopy(record)
|
|
raw = canonical(value)
|
|
try:
|
|
atomic(self.root / 'ledger.json', raw)
|
|
except BaseException:
|
|
self.anchor = None
|
|
raise
|
|
self.value, self.raw, self.anchor = value, raw, digest(raw)
|
|
os.close(self.head_fd)
|
|
self.head_fd = secure(self.root / 'ledger.json')
|
|
self.head_identity = identity(self.head_fd)
|
|
|
|
def commit_head(self, value):
|
|
"""Publish externally reconciled metadata under the existing live anchor.
|
|
|
|
No journal/consumption history may be removed by this operation.
|
|
Used for sequence-zero external reconciliation, never initialization.
|
|
"""
|
|
self.check()
|
|
for key in ('journal', 'intents', 'consumed', 'grants', 'epochs', 'readmissions'):
|
|
if value[key][:len(self.value[key])] != self.value[key]:
|
|
raise ContinuityError('history truncation')
|
|
previous_os = self.value.get('os_observations', [])
|
|
if value.get('os_observations', [])[:len(previous_os)] != previous_os or value.get('os_admission') != self.value.get('os_admission'):
|
|
raise ContinuityError('OS authority/observation history changed')
|
|
previous_checkpoints = self.value.get('os_checkpoints', [])
|
|
if value.get('os_checkpoints', [])[:len(previous_checkpoints)] != previous_checkpoints:
|
|
raise ContinuityError('OS checkpoint history truncated')
|
|
previous_manager = self.value.get('manager_events', [])
|
|
if value.get('manager_events', [])[:len(previous_manager)] != previous_manager:
|
|
raise ContinuityError('manager uncertainty history truncated')
|
|
if value['sequence'] != self.value['sequence'] or value['receipt'] != self.value['receipt']:
|
|
raise ContinuityError('head metadata cannot rewrite receipt')
|
|
raw = canonical(value)
|
|
try:
|
|
atomic(self.root / 'ledger.json', raw)
|
|
except BaseException:
|
|
self.anchor = None
|
|
raise
|
|
self.value, self.raw, self.anchor = copy.deepcopy(value), raw, digest(raw)
|
|
os.close(self.head_fd)
|
|
self.head_fd = secure(self.root / 'ledger.json')
|
|
self.head_identity = identity(self.head_fd)
|
|
|
|
def checkpoint_os(self, status):
|
|
self.check()
|
|
spec = self.binding['os_update']
|
|
if status['authority_digest'] != spec['authority_digest'] or status['plan_sha256'] != spec['plan_sha256'] or status['sequence'] != len(status['events']):
|
|
raise ContinuityError('OS checkpoint binding')
|
|
history = [x for x in self.value.get('os_checkpoints', []) if x['authority_digest'] == spec['authority_digest']]
|
|
expected = history[-1]['sequence'] + 1 if history else 1
|
|
if status['sequence'] != expected:
|
|
raise ContinuityError('OS sequence replay/gap')
|
|
if not status['events'] or status['phase'] != status['events'][-1]['phase'] or (not history and status['phase'] != 'PREPARE'):
|
|
raise ContinuityError('OS checkpoint phase')
|
|
if history:
|
|
allowed = {'PREPARE': ('DOWNLOAD', 'RECONCILIATION_REQUIRED'),
|
|
'DOWNLOAD': ('VERIFIED', 'RECONCILIATION_REQUIRED'),
|
|
'VERIFIED': ('APPLY_INTENT', 'RECONCILIATION_REQUIRED'),
|
|
'APPLY_INTENT': ('VERIFY', 'RECONCILIATION_REQUIRED'),
|
|
'VERIFY': ('REBOOT_REQUIRED', 'RECONCILIATION_REQUIRED'),
|
|
'REBOOT_REQUIRED': ('POSTBOOT_CONFIRMED',),
|
|
'RECONCILIATION_REQUIRED': ('RECONCILIATION_REQUIRED',), 'POSTBOOT_CONFIRMED': ()}
|
|
if status['phase'] not in allowed.get(history[-1]['phase'], ()):
|
|
raise ContinuityError('OS checkpoint transition')
|
|
previous = copy.deepcopy(status)
|
|
previous['events'].pop()
|
|
previous['sequence'] -= 1
|
|
previous['phase'] = previous['events'][-1]['phase']
|
|
if digest(canonical(previous)) != history[-1]['status_sha256']:
|
|
raise ContinuityError('OS checkpoint preimage changed')
|
|
value = copy.deepcopy(self.value)
|
|
value.setdefault('os_checkpoints', []).append(dict(authority_digest=spec['authority_digest'],
|
|
sequence=status['sequence'], phase=status['phase'], status_sha256=digest(canonical(status))))
|
|
self.commit_head(value)
|
|
|
|
def commit_successor(self, binding, record_digest, prerequisite, os_admission=None):
|
|
"""One atomic target-index movement; every predecessor byte is retained."""
|
|
self.check()
|
|
value = copy.deepcopy(self.value)
|
|
previous = value['receipt']
|
|
if not previous or previous['outcome'] != 'COMPLETE' or value['inbox']:
|
|
raise ContinuityError('nonterminal predecessor')
|
|
value.setdefault('predecessors', {})[value['transaction_id']] = dict(
|
|
binding=self.binding, receipt=previous, execution=value['execution'],
|
|
prerequisite=value['prerequisite'], successor_record_digest=record_digest,
|
|
os_admission=value.get('os_admission'))
|
|
value.update(active_binding=binding, transaction_id=binding['transaction_id'],
|
|
receipt=None, intent=None, execution=None, activation=None,
|
|
prerequisite=prerequisite, epoch=binding['admission_epoch_id'],
|
|
notice_grant=None, execution_grant=None)
|
|
value['os_admission'] = os_admission
|
|
value['epochs'].append(binding['admission_epoch_id'])
|
|
value['grants'].append(binding['prerequisite']['grant_digest'])
|
|
value.setdefault('successions', []).append(record_digest)
|
|
raw = canonical(value)
|
|
try:
|
|
atomic(self.root / 'ledger.json', raw)
|
|
except BaseException:
|
|
self.anchor = None
|
|
raise
|
|
self.binding = binding
|
|
self.value, self.raw, self.anchor = value, raw, digest(raw)
|
|
os.close(self.head_fd)
|
|
self.head_fd = secure(self.root / 'ledger.json')
|
|
self.head_identity = identity(self.head_fd)
|