#!/usr/bin/env python3 """ PRE_HERMES Control Plane — CP1 fail-closed Control-State Reader. Deterministischer, minimaler Reader. KEINE generische Config-Engine. Erlaubte Werte strikt definiert: "ON" | "OFF". Fail-closed Invarianten: - positive Enable-Gate: RAW != "ON" ODER fehlend/malformed/epoch-invalid -> EFFECTIVE=OFF - Emergency Stop (negativ): RAW == "ON" ODER fehlend/malformed -> EFFECTIVE=ON (restriktiver) - UNKNOWN = MORE RESTRICTIVE (niemals weniger restriktiv) Boot-Binding: - Positive ON-Grants tragen ein boot_id-Feld (eigene Zeile "boot_id="). - Grant ist nur gültig, wenn grant_boot_id == current_boot_id. - Boot-ID-Quelle: /proc/sys/kernel/random/boot_id (Kernel-Boot-ID). CP1: KEINE zeitabhängigen Leases. Boot-Binding genügt. """ import os import sys import time from datetime import datetime, timezone # Konfiguration STATE_DIR = os.environ.get("C5_CONTROL_STATE_DIR", "/opt/control-plane/state") BOOT_ID_FILE = os.environ.get("C5_BOOT_ID_FILE", "/proc/sys/kernel/random/boot_id") # Erlaubte RAW-Werte ON = "ON" OFF = "OFF" _ALLOWED = {ON, OFF} # Die sechs Controls POSITIVE_GATES = ( "global_autonomy", "productive_mutations", "save_execution", "delete_execution", "trading_execution", ) EMERGENCY = "emergency_stop" def _now_iso(): return datetime.now(timezone.utc).isoformat() def read_boot_id(): """Liest die aktuelle Kernel-Boot-ID. Fail-closed: fehlend/leer -> None.""" try: with open(BOOT_ID_FILE, "r") as f: bid = f.read().strip() # Kernel-Boot-ID ist ein UUID. Mindestplausibilität: nicht leer, keine Leerzeichen/Newlines drin. if not bid or any(ch.isspace() for ch in bid) or len(bid) < 8: return None return bid except (OSError, IOError): return None def _parse_raw(path): """ Liest eine Control-Datei. Rückgabe: raw_state : "ON" | "OFF" | None (None = fehlend/malformed) boot_id : str | None (aus boot_id-Zeile, nur relevant für positive gates) Fail-closed: unlesbar -> (None, None). """ boot_id = None try: with open(path, "r") as f: content = f.read() except (OSError, IOError): return None, None if not content or not content.strip(): return None, None # empty file -> UNKNOWN -> restriktiver lines = content.strip().splitlines() if len(lines) > 2: return None, None # malformed: zu viele Zeilen raw_state = None for line in lines: line = line.strip() if not line: continue if line in _ALLOWED: if raw_state is not None: return None, None # doppelter Wert -> malformed raw_state = line elif line.startswith("boot_id="): bid = line[len("boot_id="):].strip() if not bid or any(ch.isspace() for ch in bid): return None, None # malformed boot_id boot_id = bid else: return None, None # unknown field -> malformed if raw_state is None: return None, None # kein gültiger Wert -> malformed return raw_state, boot_id def _positive_effective(raw, grant_boot_id, current_boot_id, emergency_effective): """ Effektiver Zustand eines positiven Enable-Gates. OFF wenn: emergency==ON, raw!=ON, fehlend/malformed, oder Boot-Binding inkonsistent. """ if emergency_effective == ON: return OFF if raw != ON: return OFF # Boot-Binding: ein positiver ON-Grant MUSS eine boot_id tragen, die == current ist. if current_boot_id is None: return OFF # missing boot_id -> OFF if grant_boot_id is None: return OFF # ON-Grant ohne Boot-Binding -> ungültig -> OFF if grant_boot_id != current_boot_id: return OFF # alter Grant (andere Boot-ID) -> OFF return ON def read_control_state(): """ Berechnet den vollständigen Control-State (RAW + EFFECTIVE für alle 6 Controls). Rückgabe: dict mit allen Feldern für die Status-Projection. """ current_boot_id = read_boot_id() # 1) Emergency Stop (negativ): fail-closed -> fehlend/malformed => ON emergency_raw, _ = _parse_raw(os.path.join(STATE_DIR, EMERGENCY)) emergency_effective = ON if emergency_raw != OFF else OFF result = { "status_timestamp": _now_iso(), "current_boot_id": current_boot_id, } result["emergency_raw"] = emergency_raw if emergency_raw is not None else "MISSING" result["emergency_effective"] = emergency_effective # Kurznamen für Status-Projection (save_effective, delete_effective, ...) _short = { "global_autonomy": "global_autonomy", "productive_mutations": "productive_mutations", "save_execution": "save", "delete_execution": "delete", "trading_execution": "trading", } # 2) Positive Gates (Boot-gebunden, hierarchisch) for gate in POSITIVE_GATES: raw, grant_boot = _parse_raw(os.path.join(STATE_DIR, gate)) short = _short[gate] # Abhängigkeit: SAVE/DELETE/TRADING brauchen zusätzlich die Master-Gates (siehe unten). eff = _positive_effective(raw, grant_boot, current_boot_id, emergency_effective) result[f"{short}_raw"] = raw if raw is not None else "MISSING" result[f"{short}_effective"] = eff # 3) Hierarchische Berechnung (MISSIONS: Mutationen, SAVE, DELETE, TRADING) global_eff = result["global_autonomy_effective"] mutations_eff = result["productive_mutations_effective"] # PRODUCTIVE_MUTATIONS hängt vom Master-Gate GLOBAL_AUTONOMY ab if mutations_eff == ON and global_eff == OFF: result["productive_mutations_effective"] = OFF # SAVE / DELETE / TRADING brauchen: GLOBAL=ON AND MUTATIONS=ON AND eigener Grant gültig for gate in ("save_execution", "delete_execution", "trading_execution"): short = _short[gate] raw = result[f"{short}_raw"] grant_boot = _parse_raw(os.path.join(STATE_DIR, gate))[1] own_eff = _positive_effective(raw, grant_boot, current_boot_id, emergency_effective) if not (global_eff == ON and mutations_eff == ON and own_eff == ON): result[f"{short}_effective"] = OFF return result def format_projection(state): """Stabile, deterministische Textausgabe (read-only Status-Projection).""" keys = [ "current_boot_id", "status_timestamp", "global_autonomy_raw", "global_autonomy_effective", "productive_mutations_raw", "productive_mutations_effective", "save_raw", "save_effective", "delete_raw", "delete_effective", "trading_raw", "trading_effective", "emergency_raw", "emergency_effective", ] lines = [] for k in keys: v = state.get(k, "MISSING") if v is None: v = "MISSING" lines.append(f"{k}={v}") return "\n".join(lines) + "\n" def main(): state = read_control_state() sys.stdout.write(format_projection(state)) if __name__ == "__main__": main()