diff --git a/red-queen-architecture/control-plane/README.md b/red-queen-architecture/control-plane/README.md new file mode 100644 index 0000000..fcb66fa --- /dev/null +++ b/red-queen-architecture/control-plane/README.md @@ -0,0 +1,112 @@ +# PRE_HERMES Control Plane — Betriebsvertrag (CP1) + +Autoritative SoT für die produktive PRE_HERMES Autonomy Control Plane. +Dieses Verzeichnis ist die **einzige** Quelle für den produktiven Control-Plane-Code. +Produktiver Deploy-Pfad: `red-queen-architecture/control-plane/` → `/opt/control-plane/`. + +## 1. Zweck + +Minimale, fail-closed Sicherheitsgrundlage für zukünftige Red-Queen/Hermes-Autonomie. +**CP1 ist KEINE Autonomie-Aktivierung.** Nach CP1 ist das System mindestens so restriktiv wie vorher. + +## 2. Komponenten + +| Datei | Zweck | +|---|---| +| `control_reader.py` | Fail-closed, deterministischer Control-State-Reader (produktiver Code) | +| `test_control_reader.py` | Statische + adversariale Tests (A–J, 25 Fälle) | +| `deploy_control_plane.sh` | Reproduzierbarer Deployment-Pfad (root, idempotent) | +| `README.md` | Dieser Betriebsvertrag | + +## 3. Control State (Source of Truth) + +Produktiver Pfad: `/opt/control-plane/state/` (root:root, 0700, Dateien 0600). + +| Datei | Typ | Erlaubte RAW-Werte | +|---|---|---| +| `global_autonomy` | positiv | `ON` / `OFF` (+ `boot_id=`) | +| `productive_mutations` | positiv | `ON` / `OFF` (+ `boot_id=`) | +| `save_execution` | positiv | `ON` / `OFF` (+ `boot_id=`) | +| `delete_execution` | positiv | `ON` / `OFF` (+ `boot_id=`) | +| `trading_execution` | positiv | `ON` / `OFF` (+ `boot_id=`) | +| `emergency_stop` | negativ | `ON` / `OFF` | + +**Positives ON-Grant-Format:** +``` +ON +boot_id= +``` + +## 4. Gate-Semantik (fail-closed) + +- **Positives Gate:** `EFFECTIVE=ON` nur wenn `RAW==ON` UND `grant_boot_id == current_boot_id` UND `EMERGENCY_EFFECTIVE==OFF`. Sonst `OFF`. +- **Emergency Stop (negativ):** `EFFECTIVE=ON` wenn `RAW==ON` ODER Datei fehlt ODER malformed. `OFF` nur bei explizitem `RAW==OFF`. +- **UNKNOWN = MORE RESTRICTIVE.** Fehlende/unlesbare/malformed Datei → positives Gate `OFF`, Emergency `ON`. +- **Hierarchie:** `SAVE/DELETE/TRADING_EFFECTIVE` benötigen `GLOBAL_AUTONOMY_EFFECTIVE==ON` UND `MUTATIONS_EFFECTIVE==ON` UND eigenen Grant gültig UND `EMERGENCY_EFFECTIVE==OFF`. Kein spezifischer Execution-Switch umgeht den globalen Mutation-Gate. + +## 5. Boot-ID-Validierung + +- Quelle: `/proc/sys/kernel/random/boot_id` (Kernel-Boot-ID, ändert sich bei jedem Host-Boot). +- Positiver Grant ist nur gültig, wenn `grant_boot_id == current_boot_id`. +- `missing boot_id` → `EFFECTIVE=OFF`. `malformed boot_id` → `EFFECTIVE=OFF`. +- **Keine manuell gepflegte Epoch, kein persistenter Alt-Grant.** + +## 6. Restart-Semantik (korrekt dokumentiert) + +- **HOST REBOOT:** `current_boot_id` ändert sich → alle alten positiven Grants ungültig → `EFFECTIVE=OFF`. Operator muss erneut autorisieren. +- **CONTAINER RESTART OHNE HOST REBOOT:** `current_boot_id` bleibt gleich → ein gültiger positiver Grant kann weiterhin gültig bleiben (kein neuer Boot, keine neue Autorisierung nötig). +- **Negative/Deny-States** (`emergency_stop=ON`, `*_execution=OFF`) persistieren über Restart hinweg (fail-closed). + +## 7. Status Projection Contract + +- Produktiver Pfad: `/opt/control-plane/status/latest.txt` (root:root, 0444, read-only). +- **STATUS PROJECTION != AUTHORITY.** +- **Authority ist ausschließlich:** Control State + aktuelle Kernel-Boot-ID + fail-closed effective-state evaluation. +- `latest.txt` dient **ausschließlich der Observability**. +- Wenn die Projection fehlt/stale/malformed ist, darf das **niemals eine positive Freigabe erzeugen** (die Authority-Evaluation ist davon unabhängig). + +## 8. Ownership / Permissions + +- Control State: root:root, 0700 (dir), 0600 (Dateien). +- Status Projection: root:root, 0444. +- Audit: root:root, `chattr +a` (append-only). +- **Red Queen:** kein Host-Root, kein Docker-Socket, kein Schreibzugriff auf `/opt/control-plane`. RQ kann State nicht lesen (0700) und nicht schreiben. RQ erhält später nur read-only Status-Projection (Observability). + +## 9. Audit-Foundation + +- Pfad: `/opt/control-plane/audit/control_audit.log` (root:root, append-only `+a`). +- Format: `WHEN | WHO | WHAT | OLD | NEW | WHY | BOOT_ID`. +- RQ kann Audit nicht überschreiben (root-owned, append-only). + +## 10. Deployment (reproduzierbar) + +```bash +# Als root auf dem Host, aus frischem Forgejo-Checkout: +cd red-queen-architecture/control-plane +./deploy_control_plane.sh +``` + +- Reproduziert ausschließlich die CP1-Artefakte. +- **Überschreibt KEINE bestehenden State-Dateien** (Runtime-State bleibt unangetastet). +- Rollback: `rm -rf /opt/control-plane`. + +## 11. NICHT in Git + +- Aktuelle produktive State-Werte als Authority (State ist Runtime, nicht SoT). +- Aktuelle `boot_id` als persistente Authority. +- Secrets, Tokens, private Keys. +- Runtime Audit Log. + +## 12. Sicherheitszustand (CP1 initial) + +``` +GLOBAL_AUTONOMY_EFFECTIVE=OFF +MUTATIONS_EFFECTIVE=OFF +SAVE_EFFECTIVE=OFF +DELETE_EFFECTIVE=OFF +TRADING_EFFECTIVE=OFF +EMERGENCY_EFFECTIVE=ON +A2-A5=NOT_PRESENT HEARTBEAT=NOT_PRESENT SCHEDULER=NOT_PRESENT +AUTH.4D=FROZEN P15=FALSE DELETE_CANARY=FALSE +TRADING_ENABLED=false +``` diff --git a/red-queen-architecture/control-plane/control_reader.py b/red-queen-architecture/control-plane/control_reader.py new file mode 100755 index 0000000..0e0a5c4 --- /dev/null +++ b/red-queen-architecture/control-plane/control_reader.py @@ -0,0 +1,214 @@ +#!/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() diff --git a/red-queen-architecture/control-plane/deploy_control_plane.sh b/red-queen-architecture/control-plane/deploy_control_plane.sh new file mode 100755 index 0000000..cc9c6ef --- /dev/null +++ b/red-queen-architecture/control-plane/deploy_control_plane.sh @@ -0,0 +1,42 @@ +#!/usr/bin/env bash +# PRE_HERMES Control Plane — CP1 Deployment-Skript (reproduzierbar aus SoT) +# +# Reproduziert AUSSCHLIESSLICH die bereits vorhandenen CP1-Artefakte. +# KEINE Runtime-State-Änderung. Deployment-Code und Runtime-State strikt getrennt. +# +# Verwendung (als root auf dem Host): +# ./deploy_control_plane.sh +# +# Dieses Skript: +# - installiert control_reader.py nach /opt/control-plane/bin/ +# - erzeugt die Verzeichnisstruktur (falls nicht vorhanden) +# - überschreibt KEINE bestehenden State-Dateien (Runtime-State bleibt unangetastet) +# - erzeugt die Status-Projection neu (read-only, root-owned) +# +# Rollback: rm -rf /opt/control-plane (reine neue Dateien, keine bestehende Komponente) + +set -euo pipefail + +CP_ROOT="/opt/control-plane" +CP_BIN="$CP_ROOT/bin" +CP_STATE="$CP_ROOT/state" +CP_AUDIT="$CP_ROOT/audit" +CP_STATUS="$CP_ROOT/status" + +# 1) Verzeichnisstruktur (idempotent) +install -d -o root -g root -m 0755 "$CP_ROOT" +install -d -o root -g root -m 0700 "$CP_STATE" +install -d -o root -g root -m 0700 "$CP_AUDIT" +install -d -o root -g root -m 0755 "$CP_BIN" +install -d -o root -g root -m 0755 "$CP_STATUS" + +# 2) Reader installieren (exakt der SoT-Code) +install -o root -g root -m 0755 "$(dirname "$0")/control_reader.py" "$CP_BIN/control_reader.py" + +# 3) Status-Projection neu erzeugen (read-only, root-owned) +"$CP_BIN/control_reader.py" > "$CP_STATUS/latest.txt" +chown root:root "$CP_STATUS/latest.txt" +chmod 0444 "$CP_STATUS/latest.txt" + +echo "CP1 deployt. State-Dateien wurden NICHT überschrieben (Runtime-State bleibt erhalten)." +echo "Status-Projection: $CP_STATUS/latest.txt" diff --git a/red-queen-architecture/control-plane/test_control_reader.py b/red-queen-architecture/control-plane/test_control_reader.py new file mode 100644 index 0000000..ead6570 --- /dev/null +++ b/red-queen-architecture/control-plane/test_control_reader.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""CP1 statische + adversariale Tests (A-J) für control_reader.py.""" +import os +import shutil +import sys +import tempfile + +import control_reader as cr + +PASS = 0 +FAIL = 0 +results = [] + +def check(name, cond): + global PASS, FAIL + if cond: + PASS += 1 + results.append(f"PASS {name}") + else: + FAIL += 1 + results.append(f"FAIL {name}") + +class Env: + def __init__(self): + self.tmp = tempfile.mkdtemp() + self.state_dir = os.path.join(self.tmp, "state") + os.makedirs(self.state_dir, exist_ok=True) + # aktuelle boot_id simulieren + self.cur_boot = "boot-test-1234" + self.boot_file = os.path.join(self.tmp, "boot_id") + with open(self.boot_file, "w") as f: + f.write(self.cur_boot + "\n") + cr.STATE_DIR = self.state_dir + cr.BOOT_ID_FILE = self.boot_file + + def write(self, name, content): + with open(os.path.join(self.state_dir, name), "w") as f: + f.write(content) + + def missing_boot(self): + os.remove(self.boot_file) + +def read(): + return cr.read_control_state() + +def fmt(eff): + return {k: v for k, v in eff.items() if k.endswith("_effective")} + +# ---------- TEST 1: initial safe state ---------- +print("== TEST 1: Initial Safe State ==") +e = Env() +e.write("global_autonomy", "OFF") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "ON") +s = read() +check("initial global=OFF", s["global_autonomy_effective"] == "OFF") +check("initial mutations=OFF", s["productive_mutations_effective"] == "OFF") +check("initial save=OFF", s["save_effective"] == "OFF") +check("initial delete=OFF", s["delete_effective"] == "OFF") +check("initial trading=OFF", s["trading_effective"] == "OFF") +check("initial emergency=ON", s["emergency_effective"] == "ON") + +# ---------- TEST A: positive ON mit falscher Boot-ID -> OFF ---------- +print("== TEST A: wrong boot-id -> OFF ==") +e = Env() +e.write("global_autonomy", "ON\nboot_id=other-boot\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("A global OFF bei falscher boot_id", s["global_autonomy_effective"] == "OFF") + +# ---------- TEST B: ON aktueller boot_id aber emergency ON -> OFF ---------- +print("== TEST B: emergency ON blockiert positives Gate ==") +e = Env() +e.write("global_autonomy", f"ON\nboot_id={e.cur_boot}\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "ON") +s = read() +check("B global OFF bei emergency ON", s["global_autonomy_effective"] == "OFF") +check("B emergency ON", s["emergency_effective"] == "ON") + +# ---------- TEST C: save ON aber mutations OFF -> SAVE OFF ---------- +print("== TEST C: save ON, mutations OFF -> SAVE OFF ==") +e = Env() +e.write("global_autonomy", f"ON\nboot_id={e.cur_boot}\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", f"ON\nboot_id={e.cur_boot}\n") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("C save OFF bei mutations OFF", s["save_effective"] == "OFF") +check("C global ON", s["global_autonomy_effective"] == "ON") + +# ---------- TEST D: delete ON aber global OFF -> DELETE OFF ---------- +print("== TEST D: delete ON, global OFF -> DELETE OFF ==") +e = Env() +e.write("global_autonomy", "OFF") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", f"ON\nboot_id={e.cur_boot}\n") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("D delete OFF bei global OFF", s["delete_effective"] == "OFF") +check("D global OFF", s["global_autonomy_effective"] == "OFF") + +# ---------- TEST E: trading ON aber emergency ON -> TRADING OFF ---------- +print("== TEST E: trading ON, emergency ON -> TRADING OFF ==") +e = Env() +e.write("global_autonomy", "OFF") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", f"ON\nboot_id={e.cur_boot}\n") +e.write("emergency_stop", "ON") +s = read() +check("E trading OFF bei emergency ON", s["trading_effective"] == "OFF") +check("E emergency ON", s["emergency_effective"] == "ON") + +# ---------- TEST F: fehlende emergency_stop -> EMERGENCY ON ---------- +print("== TEST F: missing emergency_stop -> ON ==") +e = Env() +e.write("global_autonomy", "OFF") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +# KEIN emergency_stop schreiben +s = read() +check("F emergency ON bei fehlender Datei", s["emergency_effective"] == "ON") + +# ---------- TEST G: malformed emergency_stop -> ON ---------- +print("== TEST G: malformed emergency_stop -> ON ==") +e = Env() +e.write("global_autonomy", "OFF") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "MAYBE\n") +s = read() +check("G emergency ON bei malformed", s["emergency_effective"] == "ON") + +# ---------- TEST H: alter Grant nach simuliertem Reboot -> OFF ---------- +print("== TEST H: alter Grant nach Reboot -> OFF ==") +e = Env() +e.write("global_autonomy", f"ON\nboot_id={e.cur_boot}\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("H vor Reboot global ON", s["global_autonomy_effective"] == "ON") +# simulierter Reboot: boot_id ändert sich +with open(e.boot_file, "w") as f: + f.write("boot-after-reboot-999\n") +s = read() +check("H nach Reboot global OFF", s["global_autonomy_effective"] == "OFF") + +# ---------- Missing boot_id file -> positive gates OFF ---------- +print("== TEST: missing boot_id file ==") +e = Env() +e.write("global_autonomy", f"ON\nboot_id={e.cur_boot}\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +e.missing_boot() +s = read() +check("missing boot_id -> global OFF", s["global_autonomy_effective"] == "OFF") + +# ---------- Permission/read error ---------- +print("== TEST: unreadable control file ==") +e = Env() +e.write("global_autonomy", "ON") # no boot_id -> invalid anyway +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "ON") +os.chmod(os.path.join(e.state_dir, "emergency_stop"), 0o000) +s = read() +check("unreadable emergency -> ON", s["emergency_effective"] == "ON") +os.chmod(os.path.join(e.state_dir, "emergency_stop"), 0o644) + +# ---------- Positive gate OHNE boot_id (ON-Grant ohne Binding) -> OFF ---------- +print("== TEST: ON-Grant ohne boot_id ==") +e = Env() +e.write("global_autonomy", "ON\n") # no boot_id line +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("ON ohne boot_id -> global OFF", s["global_autonomy_effective"] == "OFF") + +# ---------- Empty file -> OFF (positive), ON (emergency) ---------- +print("== TEST: empty positive file ==") +e = Env() +e.write("global_autonomy", "") # empty +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("empty positive -> global OFF", s["global_autonomy_effective"] == "OFF") +check("empty positive raw MISSING", s["global_autonomy_raw"] == "MISSING") + +# ---------- unknown field -> malformed -> OFF ---------- +print("== TEST: unknown field ==") +e = Env() +e.write("global_autonomy", "ON\nfoo=bar\nboot_id=xyz\n") +e.write("productive_mutations", "OFF") +e.write("save_execution", "OFF") +e.write("delete_execution", "OFF") +e.write("trading_execution", "OFF") +e.write("emergency_stop", "OFF") +s = read() +check("unknown field -> global OFF", s["global_autonomy_effective"] == "OFF") + +print() +print(f"RESULT: PASS={PASS} FAIL={FAIL}") +for r in results: + print(r) +sys.exit(1 if FAIL else 0)