"""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)