trading-system-docs/red-queen-architecture/control-plane/control_reader.py

#!/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=<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 fünf Controls (Hermes/Red-Queen-Domäne; Trading ist eine separate Security-Domain)
POSITIVE_GATES = (
    "global_autonomy",
    "productive_mutations",
    "save_execution",
    "delete_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 5 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",
    }

    # 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 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)
    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 brauchen: GLOBAL=ON AND MUTATIONS=ON AND eigener Grant gültig
    for gate in ("save_execution", "delete_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",
        "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()