"""Local promotion evidence. Invoked by the root RA; never invokes cluster tools. State is a bounded JSON record, never executable input. The public CLI uses a fixed root-owned /run directory and CLOCK_BOOTTIME, including suspend duration. """ import fcntl import hashlib import json import math import os from pathlib import Path import re import stat import sys import tempfile import time STATE_DIR = Path('/run/managed-container/promotion') def unique_object(pairs): result = {} for key, value in pairs: if key in result: raise ValueError('duplicate JSON key') result[key] = value return result def lsn(value): if not isinstance(value, str) or not re.fullmatch(r'[0-9A-F]{1,8}/[0-9A-F]{1,8}', value): raise ValueError('invalid LSN') high, low = value.split('/') return (int(high, 16) << 32) | int(low, 16) def number(value): return type(value) in (int, float) and math.isfinite(value) and value >= 0 def validate_observation(obs): if set(obs) != {'identity', 'recovery', 'paused', 'receiver', 'replay', 'flush', 'strict'}: raise ValueError('invalid observation fields') identity = obs['identity'] if not isinstance(identity, list) or len(identity) != 3: raise ValueError('invalid identity') system, path, started = identity if (not isinstance(system, str) or not re.fullmatch(r'[0-9]{1,20}', system) or not isinstance(path, str) or not path.startswith('/') or len(path) > 2048 or not isinstance(started, str) or not re.fullmatch(r'[0-9]{1,12}\.[0-9]{1,6}', started)): raise ValueError('invalid identity values') if type(obs['recovery']) is not bool or type(obs['paused']) is not bool: raise ValueError('invalid role') if obs['strict'] is not None and type(obs['strict']) is not bool: raise ValueError('invalid strict flag') if obs['receiver'] is not None and obs['receiver'] not in ( 'starting', 'streaming', 'waiting', 'restarting', 'stopping'): raise ValueError('invalid receiver status') replay = lsn(obs['replay']) if obs['replay'] is not None else None flush = lsn(obs['flush']) if obs['flush'] is not None else None strict = (obs['recovery'] and not obs['paused'] and obs['receiver'] == 'streaming' and replay is not None and flush is not None and replay >= flush) if (obs['strict'] is True) != strict: raise ValueError('inconsistent strict observation') return strict class Evidence: def __init__(self, directory, context, boot, now, wall, owner=0): self.directory, self.context = Path(directory), context self.boot, self.now, self.wall, self.owner = boot, now, wall, owner key = hashlib.sha256(json.dumps(context, separators=(',', ':')).encode()).hexdigest() self.path = self.directory / (key + '.json') self.lock = self.directory / (key + '.lock') def secure(self, path, mode): info = path.lstat() if (info.st_uid != self.owner or stat.S_IMODE(info.st_mode) != mode or (not stat.S_ISREG(info.st_mode) if mode == 0o600 else not stat.S_ISDIR(info.st_mode))): raise ValueError('unsafe state ownership/type/mode') def discard(self): self.path.unlink(missing_ok=True) def evaluate(self, obs, max_age): if type(max_age) is not int or not 0 <= max_age <= 99999: raise ValueError('invalid freshness threshold') self.directory.mkdir(mode=0o700, exist_ok=True) self.secure(self.directory, 0o700) fd = os.open(self.lock, os.O_CREAT | os.O_NOFOLLOW | os.O_RDWR, 0o600) with os.fdopen(fd, 'r+') as lock: self.secure(self.lock, 0o600) fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) try: return self._evaluate(obs, max_age) except (ValueError, TypeError, KeyError, OverflowError): self.discard() return '-INFINITY' def _evaluate(self, obs, max_age): strict = validate_observation(obs) if strict: record = dict(version=1, context=self.context, boot=self.boot, identity=obs['identity'], observed_boottime=self.now, observed_unix=self.wall, replay=obs['replay'], flush=obs['flush']) # Same-directory atomic replace; failed persistence cannot authorize # subsequent disconnected promotion. No secrets are recorded. fd, temporary = tempfile.mkstemp(prefix='.observation-', dir=self.directory) try: with os.fdopen(fd, 'w') as out: json.dump(record, out, separators=(',', ':')) out.flush() os.fsync(out.fileno()) os.replace(temporary, self.path) finally: Path(temporary).unlink(missing_ok=True) return '100' if (not obs['recovery'] or obs['paused'] or obs['receiver'] == 'streaming' or obs['replay'] is None or max_age == 0): self.discard() return '-INFINITY' if not self.path.exists(): return '-INFINITY' self.secure(self.path, 0o600) if self.path.stat().st_size > 4096: raise ValueError('oversized state') with self.path.open() as source: record = json.load(source, object_pairs_hook=unique_object) if set(record) != {'version', 'context', 'boot', 'identity', 'observed_boottime', 'observed_unix', 'replay', 'flush'}: raise ValueError('invalid record fields') if (type(record['version']) is not int or record['version'] != 1 or record['context'] != self.context or record['boot'] != self.boot or record['identity'] != obs['identity'] or not number(record['observed_boottime']) or not number(record['observed_unix']) or not 0 <= self.now - record['observed_boottime'] <= max_age or lsn(record['replay']) < lsn(record['flush']) or lsn(obs['replay']) < lsn(record['replay'])): raise ValueError('expired or incompatible observation') return '50' def main(): if os.geteuid() != 0 or len(sys.argv) != 6 or not all(sys.argv[1:5]): raise ValueError('root and explicit resource context required') age = sys.argv[5] if not re.fullmatch(r'(0|[1-9][0-9]{0,4})', age): raise ValueError('invalid freshness threshold') # /run is trusted root-owned; validate our parent before using a child. parent = STATE_DIR.parent parent.mkdir(mode=0o700, exist_ok=True) info = parent.lstat() if info.st_uid != 0 or not stat.S_ISDIR(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o700: raise ValueError('unsafe state parent') raw = sys.stdin.read(8193) if len(raw) > 8192: raise ValueError('oversized observation') evidence = Evidence(STATE_DIR, sys.argv[1:5], Path('/proc/sys/kernel/random/boot_id').read_text().strip(), time.clock_gettime(time.CLOCK_BOOTTIME), time.time()) try: obs = json.loads(raw, object_pairs_hook=unique_object) except ValueError: obs = {} # Invalid query output withdraws and removes prior evidence. print(evidence.evaluate(obs, int(age))) if __name__ == '__main__': try: main() except (OSError, ValueError, TypeError, KeyError, OverflowError): print('-INFINITY') sys.exit(1)