#!/usr/bin/python3 -I """Single-use NTS recovery handoff for the protected C01 supervisor. The root supervisor calls prepare() BEFORE starting the authority's challenge exchange. Its dedicated child permanently drops to the authority's administrative UID, creates the admin endpoint under that identity, then calls send_once(). Only an inherited in-memory observation is accepted; there is no saved-report, JSON-input, command-line or raw-clock recovery path. PlatformQualification is supervisor-supplied evidence, not something this module can manufacture from a current clock read. The supervisor must qualify the actual platform's suspend/clock behavior, protect this code and its policy, pin the child executables, exclude old writers and own both processes for their whole lifetime. This module does not install those controls or assert installed acceptance. """ from dataclasses import dataclass from decimal import ROUND_FLOOR import ctypes import hashlib import json import os from pathlib import Path import select import socket import struct import time import c01_nts as nts NANO = 1_000_000_000 IO_BUDGET_NS = 2 * NANO PACKET_BYTES = 8192 @dataclass(frozen=True) class Peer: pid: int uid: int gid: int @dataclass(frozen=True) class PlatformQualification: # Exact platform and evidence identity, supplied by protected provisioning. kernel_release: str clocksource: str evidence_sha256: str max_bracket_ns: int max_observation_error_ns: int max_elapsed_error_ppb: int max_awake_ns: int = None characterized_model: dict | None = None def validate(self): nts.require(self.kernel_release == os.uname().release and self.clocksource == clocksource(), 'recovery platform qualification mismatch') nts.require(type(self.evidence_sha256) is str and len(self.evidence_sha256) == 64 and all(c in '0123456789abcdef' for c in self.evidence_sha256) and self.evidence_sha256 != '0' * 64, 'missing platform qualification evidence') for value in (self.max_bracket_ns, self.max_observation_error_ns): nts.require(type(value) is int and 0 < value <= (2 ** 64 - 1) * NANO, 'invalid platform qualification bound') nts.require(type(self.max_elapsed_error_ppb) is int and 0 <= self.max_elapsed_error_ppb <= 2 ** 64 - 1, 'invalid qualified elapsed-time error bound') def validate_epoch(self, maximum): self.validate() nts.require(type(self.max_awake_ns) is int and self.max_awake_ns > 0 and type(maximum) is int and maximum == self.max_awake_ns, 'qualified awake lifetime unavailable or delegation differs') if self.characterized_model is not None: validate_characterized_model(self.characterized_model) nts.require(nts.platform_identity() == self.characterized_model['identity'], 'characterized hardware/kernel/clocksource mismatch') @dataclass(frozen=True) class EpochIssuerDelegation: """Protected supervisor appointment, not an application-selected request. Automatic NTS is an authorized issuer. Wake scope must match the separately protected owner delegation; no interactive human approval is implied here. """ identity: bytes allow_wake: bool allow_refresh: bool = False def validate(self): nts.require(type(self.identity) is bytes and len(self.identity) == 32 and any(self.identity) and type(self.allow_wake) is bool and type(self.allow_refresh) is bool, 'invalid epoch issuer delegation') def duration_ns(value): nts.require(type(value) is dict and set(value) == {'secs', 'nanos'} and type(value['secs']) is int and 0 <= value['secs'] < 2**64 and type(value['nanos']) is int and 0 <= value['nanos'] < NANO, 'invalid model duration') return value['secs'] * NANO + value['nanos'] def validate_characterized_model(model): nts.require(type(model) is dict and set(model) == {'identity','evidence_sha256','additive', 'rate_numerator','rate_denominator','max_raw_holdover','max_reference_age','max_adjustment','association_multiplier'}, 'invalid characterized model fields') identity = model['identity'] nts.require(type(identity) is dict and set(identity) == {'kernel','clocksource','hardware'}, 'invalid model identity') for value in [*identity.values(), model['evidence_sha256']]: nts.require(type(value) is list and len(value) == 32 and any(value) and all(type(n) is int and 0 <= n <= 255 for n in value), 'invalid model identity/evidence') n, d = model['rate_numerator'], model['rate_denominator'] nts.require(type(n) is int and type(d) is int and 0 <= n < d < 2**64, 'invalid characterized rate') for name in ('additive','max_raw_holdover','max_reference_age','max_adjustment'): duration_ns(model[name]) nts.require(duration_ns(model['max_raw_holdover']) > 0 and duration_ns(model['max_reference_age']) > 0 and type(model['association_multiplier']) is int and 0 < model['association_multiplier'] < 2**32, 'invalid model horizon/association') def model_reference_age(model, observation, reference): age = observation.utc_ns - reference + observation.observation_error_ns nts.require(0 <= age <= duration_ns(model['max_reference_age']), 'characterized reference too old') n, d = model['rate_numerator'], model['rate_denominator'] return ((age + duration_ns(model['additive'])) * d + d - n - 1) // (d - n) def model_error(model, raw_anchor, anchor_boot, raw, before, after, anchor_error, reference_age, barrier=False): validate_characterized_model(model) for witness in (raw_anchor, raw): nts.require(type(witness) is dict and set(witness) == {'before','after','identity'} and witness['identity'] == model['identity'] and duration_ns(witness['after']) >= duration_ns(witness['before']), 'missing/foreign RAW witness') a, b = duration_ns(raw_anchor['before']), duration_ns(raw_anchor['after']) lo, hi = duration_ns(raw['before']), duration_ns(raw['after']) nts.require(lo >= a and hi >= b and after >= before >= anchor_boot, 'RAW/BOOTTIME regression') low, high = max(0, lo - b), hi - a total = high + reference_age if not barrier: nts.require(total < duration_ns(model['max_raw_holdover']), 'RAW holdover expired') disagreement = max(abs(before - anchor_boot - low), abs(before - anchor_boot - high)) if not barrier: nts.require(disagreement <= duration_ns(model['max_adjustment']), 'cumulative BOOTTIME/RAW adjustment refused') n, d = model['rate_numerator'], model['rate_denominator'] chunks=max(1,(total+duration_ns(model['max_raw_holdover'])-1)//duration_ns(model['max_raw_holdover'])) if barrier else 1 return anchor_error + chunks*duration_ns(model['additive']) + (total*n+d-1)//d + chunks-1 + disagreement + (after-before)*model['association_multiplier'] def epoch_max_awake_ns(challenge): nts.require(type(challenge) is dict and type(challenge.get('version')) is int and challenge['version'] == 3, 'epoch characterized/expiry protocol required') value = challenge.get('max_awake') nts.require(type(value) is dict and set(value) == {'secs', 'nanos'} and type(value['secs']) is int and 0 <= value['secs'] < 2**64 and type(value['nanos']) is int and 0 <= value['nanos'] < NANO, 'invalid epoch awake lifetime') maximum = value['secs'] * NANO + value['nanos'] nts.require(maximum > 0, 'invalid epoch awake lifetime') return maximum def epoch_packet(observation, encoded_binding, qualification, challenge, delegation): """Encode one fresh observation for an authenticated owner's epoch challenge. This pure encoder does not authenticate a socket or make a recovery grant. send_epoch_once owns those checks and the one-use PreparedRecovery lifetime. """ delegation.validate() fields = {'version', 'binding', 'delegation_id', 'application_owner', 'epoch', 'cause', 'boot_id', 'not_before', 'domain', 'nonce', 'max_awake', 'characterized', 'barrier_raw'} nts.require(type(challenge) is dict and set(challenge) == fields, 'invalid epoch challenge fields') binding = json.loads(encoded_binding) maximum = epoch_max_awake_ns(challenge) nts.require(type(qualification.max_awake_ns) is int and qualification.max_awake_ns == maximum, 'qualified awake lifetime unavailable or delegation differs') expiry = duration(observation.boottime_ns + maximum) nts.require(type(challenge['version']) is int and challenge['version'] == 3 and challenge['binding'] == binding and binding_bytes(challenge['binding']) == encoded_binding, 'foreign epoch challenge binding/version') for name, length in (('delegation_id', 32), ('application_owner', 32), ('boot_id', 16), ('domain', 16), ('nonce', 32)): value = challenge[name] nts.require(type(value) is list and len(value) == length and any(value) and all(type(n) is int and 0 <= n <= 255 for n in value), 'invalid epoch challenge identity') nts.require(bytes(challenge['delegation_id']) == delegation.identity and bytes(challenge['boot_id']).hex() == observation.boot_id, 'foreign epoch delegation/boot') nts.require(challenge['cause'] in ('Restart', 'Wake', 'Refresh') and (challenge['cause'] != 'Wake' or delegation.allow_wake) and (challenge['cause'] != 'Refresh' or delegation.allow_refresh), 'issuer delegation does not cover recovery cause') epoch = challenge['epoch'] nts.require(type(epoch) is int and 0 < epoch < 2**64, 'invalid consumed epoch') barrier = challenge['not_before'] nts.require(type(barrier) is dict and set(barrier) == {'secs', 'nanos'} and type(barrier['secs']) is int and 0 <= barrier['secs'] < 2**64 and type(barrier['nanos']) is int and 0 <= barrier['nanos'] < NANO, 'invalid epoch freshness barrier') nts.require(observation.boottime_ns >= barrier['secs'] * NANO + barrier['nanos'], 'pre-revocation/pre-wake observation refused') nts.require(isinstance(observation.assessment, nts.Assessment), 'missing authenticated reference assessment') reference = int((observation.assessment.reference_time * nts.NANO).to_integral_value(rounding=ROUND_FLOOR)) since_barrier = observation.boottime_ns - (barrier['secs'] * NANO + barrier['nanos']) model = challenge['characterized'] nts.require(model == qualification.characterized_model, 'characterized model differs from protected qualification') if model is not None: validate_characterized_model(model) error = model_error(model, challenge['barrier_raw'], duration_ns(barrier), observation.raw, observation.boottime_ns, observation.collected_at_ns, observation.observation_error_ns, model_reference_age(model, observation, reference), barrier=True) else: error = observation.observation_error_ns + (since_barrier * qualification.max_elapsed_error_ppb + NANO - 1) // NANO boundary_upper = observation.utc_ns - since_barrier + error nts.require(0 <= boundary_upper < reference <= observation.utc_ns, 'stale/pre-boundary authenticated reference') data = (b'C01 qualified epoch v1\0' + bytes(binding['local_identity']) + bytes(binding['incarnation']) + binding['reader_schema'].to_bytes(4, 'big') + binding['writer_schema'].to_bytes(4, 'big') + delegation.identity + epoch.to_bytes(8, 'big') + bytes(challenge['boot_id'])) domain = hashlib.sha256(data).digest()[:16] nts.require(bytes(challenge['domain']) == domain, 'old or foreign epoch domain') return {'version': 3, 'challenge': challenge, 'reference_wall': duration(reference), 'raw_anchor': observation.raw if model is not None else None, 'expires_at': expiry, 'recovery': { 'domain': list(domain), 'boot_id': challenge['boot_id'], 'wall': duration(observation.utc_ns), 'monotonic': duration(observation.boottime_ns), 'max_bracket': duration(qualification.max_bracket_ns), 'uncertainty': duration(observation.observation_error_ns), 'max_elapsed_error_ppb': qualification.max_elapsed_error_ppb}} def clocksource(): return Path('/sys/devices/system/clocksource/clocksource0/current_clocksource').read_text().strip() def boot_id(): value = Path('/proc/sys/kernel/random/boot_id').read_text().strip().replace('-', '') nts.require(len(value) == 32 and all(c in '0123456789abcdef' for c in value) and value != '0' * 32, 'invalid recovery boot identity') return value def duration(nanos): nts.require(type(nanos) is int and 0 <= nanos <= (2 ** 64 - 1) * NANO, 'invalid recovery duration') return {'secs': nanos // NANO, 'nanos': nanos % NANO} def binding_bytes(binding): nts.require(type(binding) is dict and set(binding) == {'local_identity', 'incarnation', 'reader_schema', 'writer_schema'}, 'invalid recovery binding fields') for name in ('local_identity', 'incarnation'): value = binding[name] nts.require(type(value) is list and len(value) == 32 and all(type(n) is int and 0 <= n <= 255 for n in value) and any(value), 'invalid recovery identity/incarnation') nts.require(type(binding['reader_schema']) is int and binding['reader_schema'] == 2 and type(binding['writer_schema']) is int and binding['writer_schema'] == 2, 'unsupported recovery schema') # Freeze the supervisor input; later mutations of its dict cannot change it. return json.dumps(binding, separators=(',', ':'), sort_keys=True).encode('ascii') def check_peer(channel, peer): nts.require(type(peer.pid) is int and peer.pid > 1 and peer.pid != os.getpid() and type(peer.uid) is int and peer.uid > 0 and type(peer.gid) is int and peer.gid > 0, 'invalid authority peer') nts.require(channel.family == socket.AF_UNIX and channel.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) == socket.SOCK_SEQPACKET, 'recovery requires a Unix packet capability') actual = struct.unpack('3i', channel.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)) nts.require(actual == (peer.pid, peer.uid, peer.gid), 'recovery authority peer mismatch') def check_administrator(parent, peer): uid, gid = os.geteuid(), os.getegid() nts.require(uid != 0 and gid != 0 and os.getresuid() == (uid, uid, uid) and os.getresgid() == (gid, gid, gid) and set(os.getgroups()) <= {gid}, 'recovery administrator must permanently drop privileges') nts.require(peer.uid == uid and peer.gid == gid, 'recovery administrator principal mismatch') status = dict(line.split(':', 1) for line in Path('/proc/self/status').read_text().splitlines() if ':' in line) nts.require(status['NoNewPrivs'].strip() == '1' and all(int(status[name], 16) == 0 for name in ('CapEff', 'CapPrm', 'CapInh', 'CapAmb')), 'recovery administrator retains elevation') nts.require(os.getppid() == parent and parent > 1, 'recovery supervisor is no longer the parent') parent_status = dict(line.split(':', 1) for line in Path(f'/proc/{parent}/status').read_text().splitlines() if ':' in line) nts.require(set(map(int, parent_status['Uid'].split())) == {0} and set(map(int, parent_status['Gid'].split())) == {0}, 'recovery supervisor is not root') # No process inspection of the inherited approval after privilege dropping. nts.require(ctypes.CDLL(None, use_errno=True).prctl(4, 0, 0, 0, 0) == 0, 'recovery administrator inspection exclusion failed') def wait_ready(channel, event, deadline): poller = select.poll() poller.register(channel, event) # A BOOTTIME deadline includes suspend; poll itself may use another clock. for _ in range(64): remaining = deadline - time.clock_gettime_ns(time.CLOCK_BOOTTIME) nts.require(remaining > 0, 'recovery challenge deadline expired') ready = poller.poll(min(50, max(1, (remaining + 999_999) // 1_000_000))) nts.require(time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery challenge deadline expired') if ready: nts.require(ready[0][1] & event and not ready[0][1] & (select.POLLERR | select.POLLNVAL), 'recovery challenge channel unavailable') return raise nts.Unavailable('recovery challenge wait exhausted') class PreparedRecovery: """One launch's root-collected observation, inherited only by its admin child.""" __slots__ = ('_observation', '_binding', '_qualification', '_domain', '_parent', '_used', '_epoch_challenge') def __init__(self, observation, binding, qualification, domain, epoch_challenge=None): self._observation = observation self._binding = binding self._qualification = qualification self._domain = domain self._parent = os.getpid() self._used = False self._epoch_challenge = epoch_challenge def __reduce_ex__(self, protocol): raise TypeError('recovery approval must not be serialized') def send_epoch_once(self, channel, peer, delegation): """Automatic epoch issuer path used by staged protected startup. Never recycles this observation after wake, failure, deadline or delivery uncertainty. The delegated service must collect anew after the owner's durable barrier. No profile or human-confirmation policy is created. """ try: nts.require(not self._used, 'recovery approval already consumed') self._used = True check_administrator(self._parent, peer) check_peer(channel, peer) self._qualification.validate() observation = self._observation deadline = observation.collected_at_ns + IO_BUDGET_NS nts.require(boot_id() == observation.boot_id and observation.collected_at_ns <= time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery observation is stale or from another boot') wait_ready(channel, select.POLLIN, deadline) raw, ancillary, flags, _ = channel.recvmsg(PACKET_BYTES, 0, socket.MSG_DONTWAIT) nts.require(raw and not ancillary and not flags, 'invalid epoch challenge packet') received = json.loads(raw) nts.require(self._epoch_challenge is None or received == self._epoch_challenge, 'issuer received a different appointed epoch/session') maximum = epoch_max_awake_ns(received) self._qualification.validate_epoch(maximum) packet = json.dumps(epoch_packet(observation, self._binding, self._qualification, received, delegation), separators=(',', ':')).encode('ascii') deadline = min(deadline, observation.boottime_ns + maximum) nts.require(len(packet) <= PACKET_BYTES, 'oversized epoch recovery authorization') self._qualification.validate() wait_ready(channel, select.POLLOUT, deadline) nts.require(boot_id() == observation.boot_id, 'recovery boot changed before send') nts.require(time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery challenge deadline expired') sent = channel.send(packet, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) nts.require(sent == len(packet) and time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'epoch recovery authorization delivery uncertain') finally: channel.close() def send_once(self, channel, peer): """Send once and close. Delivery is not proof of authority acceptance.""" # Failure, timeout, malformed challenge and uncertain send consume it too. try: nts.require(not self._used, 'recovery approval already consumed') self._used = True check_administrator(self._parent, peer) check_peer(channel, peer) self._qualification.validate() observation = self._observation deadline = observation.collected_at_ns + IO_BUDGET_NS nts.require(boot_id() == observation.boot_id and observation.collected_at_ns <= time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery observation is stale or from another boot') wait_ready(channel, select.POLLIN, deadline) challenge, ancillary, flags, _ = channel.recvmsg(32, 0, socket.MSG_DONTWAIT) nts.require(len(challenge) == 32 and not ancillary and not flags, 'invalid recovery challenge packet') # Keep the original authenticated UTC/BOOTTIME mapping. No raw-wall # substitution, watermark clamping or age-based reanchoring. packet = json.dumps({'version': 4, 'binding': json.loads(self._binding), 'challenge': list(challenge), 'recovery': { 'domain': list(self._domain), 'boot_id': list(bytes.fromhex(observation.boot_id)), 'wall': duration(observation.utc_ns), 'monotonic': duration(observation.boottime_ns), 'max_bracket': duration(self._qualification.max_bracket_ns), # collect() already includes its complete sampling span in # this error. Retain it plus the measured elapsed-time model. 'uncertainty': duration(observation.observation_error_ns), 'max_elapsed_error_ppb': self._qualification.max_elapsed_error_ppb, }}, separators=(',', ':')).encode('ascii') nts.require(len(packet) <= PACKET_BYTES, 'oversized recovery authorization') wait_ready(channel, select.POLLOUT, deadline) nts.require(boot_id() == observation.boot_id, 'recovery boot changed before send') nts.require(time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery challenge deadline expired') sent = channel.send(packet, socket.MSG_DONTWAIT | socket.MSG_NOSIGNAL) nts.require(sent == len(packet) and time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline, 'recovery authorization delivery uncertain') finally: channel.close() def prepare(policy, qualification, binding, domain): """Root-only live collection before launch; no deserialization/fallback input. domain is the independently qualified supervisor's clock incarnation, not a process ID, caller-selected request field, or a random domain generated here. A successful return does not attest old-writer exclusion or authorize launch. """ nts.require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'recovery preparation requires the root supervisor') qualification.validate() encoded = binding_bytes(binding) nts.require(type(domain) is bytes and len(domain) == 16 and any(domain), 'missing qualified recovery domain') observation = nts.collect(policy) qualification.validate() nts.require(observation.boot_id == boot_id() and 0 <= observation.sampling_bracket_ns <= qualification.max_bracket_ns and 0 <= observation.observation_error_ns <= qualification.max_observation_error_ns and observation.collected_at_ns == observation.boottime_ns + observation.sampling_bracket_ns, 'NTS observation exceeds platform qualification') duration(observation.utc_ns) duration(observation.boottime_ns) return PreparedRecovery(observation, encoded, qualification, domain) def prepare_epoch(policy, qualification, binding, challenge, delegation, max_awake_ns, deadline): """One automatic authorization attempt, fresh network reference after barrier. Waiting for chrony's quality/reference does not resend or reissue any grant. The committed challenge is immutable and the caller owns the collector until retirement. The final inherited single-use send still has its two-second age. """ nts.require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'root epoch collector required') qualification.validate_epoch(max_awake_ns) delegation.validate() encoded = binding_bytes(binding) context = json.loads(json.dumps(challenge)) nts.require(epoch_max_awake_ns(context) == max_awake_ns, 'epoch awake lifetime differs from appointment') # Validate protected context before waiting; never launder a Wake as Restart. nts.require(context['binding'] == binding and bytes(context['delegation_id']) == delegation.identity and context['cause'] in ('Restart', 'Wake', 'Refresh') and (context['cause'] != 'Wake' or delegation.allow_wake) and (context['cause'] != 'Refresh' or delegation.allow_refresh) and context['characterized'] == qualification.characterized_model, 'epoch issuer scope/model mismatch') last_refusal=None while time.clock_gettime_ns(time.CLOCK_BOOTTIME) < deadline: try: observation = nts.collect(policy, raw=True) if context['characterized'] is not None else nts.collect(policy) qualification.validate_epoch(max_awake_ns) nts.require(observation.boot_id == boot_id() and observation.sampling_bracket_ns <= qualification.max_bracket_ns and observation.observation_error_ns <= qualification.max_observation_error_ns, 'NTS observation exceeds platform qualification') epoch_packet(observation, encoded, qualification, context, delegation) except nts.Unavailable as error: reason=str(error) if reason!=last_refusal: os.write(2,('C01_EPOCH_WAIT '+reason[:256]+'\n').encode()) last_refusal=reason if bytes(context['boot_id']).hex() != boot_id(): raise time.sleep(.25) continue return PreparedRecovery(observation, encoded, qualification, bytes(context['domain']), context) raise nts.Unavailable('fresh post-barrier NTS reference unavailable before setup deadline: '+str(last_refusal))