CP2A1: Trusted Gate Evaluator (External Gate Enforcement Foundation) - non-productive, no autonomy, CP1-logic reuse, fail-closed, internal-net, no credentials

red-queen-architecture/control-plane/cp2a1/Dockerfile

@@ -0,0 +1,26 @@
+# CP2A1 — Trusted Gate Evaluator (External Gate Enforcement Foundation)
+# STRICTLY NON-PRODUCTIVE / NO AUTONOMY.
+# Minimaler Container: python:3.12-slim, kein Shell-Zugriff, read-only FS.
+#
+# HINWEIS zum User: Der Evaluator MUSS die root-owned CP1-State-Dateien
+# (/opt/control-plane/state/, root:root 0700/600) lesen. Daher laeuft der
+# Container als root (container-intern, KEIN Host-Root — konsistent mit dem
+# CP2D-Ground-Truth-Muster). Die Härtung erfolgt ueber:
+#   --cap-drop ALL --no-new-privileges --read-only --network <internal>
+#   RO-State-Mount, kein Docker-Socket, keine Credentials, keine Egress.
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# Nur die Evaluator-Dateien (kein Build-Artefakt, kein Test-Code im Image)
+COPY gate_evaluator.py /app/gate_evaluator.py
+COPY control_reader.py /app/control_reader.py
+
+# Audit-Verzeichnis (eigenes Volume, append-only gegen RQ-Manipulation)
+RUN mkdir -p /audit
+
+# Kein Shell-Entrypoint: direkt Python. Kein /bin/sh im CMD.
+ENTRYPOINT ["python3", "/app/gate_evaluator.py"]

red-queen-architecture/control-plane/cp2a1/control_reader.py

@@ -0,0 +1,210 @@
+#!/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()

red-queen-architecture/control-plane/cp2a1/docker-compose.yml

@@ -0,0 +1,38 @@
+# CP2A1 — Trusted Gate Evaluator (External Gate Enforcement Foundation)
+# STRICTLY NON-PRODUCTIVE / NO AUTONOMY.
+#
+# Netzwerk: eigenes internal-Netz (keine Egress, keine Internet-Erreichbarkeit).
+# RQ wird per `docker network connect` an dieses Netz angebunden (RQ <-> Evaluator).
+# State-Mount: RO (read-only) auf /opt/control-plane/state.
+# Härtung: --cap-drop ALL, --no-new-privileges, --read-only, kein Docker-Socket.
+services:
+  gate-evaluator:
+    build: .
+    image: gate-evaluator:cp2a1
+    container_name: gate-evaluator
+    restart: unless-stopped
+    read_only: true
+    cap_drop:
+      - ALL
+    security_opt:
+      - no-new-privileges:true
+    tmpfs:
+      - /tmp
+    volumes:
+      # RO-State-Mount: autoritativer CP1-State (root-owned, 0700/600)
+      - /opt/control-plane/state:/opt/control-plane/state:ro
+      # Audit-Volume (append-only, gegen RQ-Manipulation geschuetzt)
+      - gate-audit:/audit
+    networks:
+      - gate-internal
+    environment:
+      - GATE_EVALUATOR_PORT=8080
+    # Kein Docker-Socket, keine Credentials, keine Egress (internal-Netz)
+
+networks:
+  gate-internal:
+    driver: bridge
+    internal: true
+
+volumes:
+  gate-audit:

red-queen-architecture/control-plane/cp2a1/gate_evaluator.py

@@ -0,0 +1,314 @@
+#!/usr/bin/env python3
+"""
+PRE_HERMES CP2A1 — Trusted Gate Evaluator (External Gate Enforcement Foundation).
+
+STRICTLY NON-PRODUCTIVE / NO AUTONOMY. Reine Gate-Entscheidungs-Boundary.
+
+Security-Prinzip:
+  RED_QUEEN_DECISION != AUTHORIZATION
+  RED_QUEEN_GATE_CHECK != TRUSTED_ENFORCEMENT
+  Die Security Boundary liegt AUSSERHALB des hermes-red-queen Containers.
+
+Dieser Evaluator:
+  - liest ausschliesslich den autoritativen CP1-State (/opt/control-plane/state/*)
+  - liest current_boot_id frisch (/proc/sys/kernel/random/boot_id)
+  - verwendet die bestehende CP1-Evaluator-Logik (control_reader.py aus der SoT)
+  - berechnet den Effective-State
+  - liefert eine kleine deterministische Gate-Entscheidung (ALLOW/DENY)
+  - erzeugt Audit
+  - fuehrt KEINE produktive Aktion aus
+  - haelt KEINE mutierenden Credentials
+  - hat KEINE Executor-Integration
+
+CP1-Logik-Reuse: importiert control_reader (SoT-Modul). KEINE Copy/Paste-Drift.
+Die ENV-basierten Pfade (C5_CONTROL_STATE_DIR / C5_BOOT_ID_FILE) werden HART
+UEBERSCHRIEBEN, damit RQ den Evaluator nicht dazu bringen kann, alternative
+State-Pfade zu lesen (user-supplied state path = verboten).
+
+Fail-closed: Jede Unklarheit -> DENY. Emergency unklar = ON.
+"""
+
+import json
+import os
+import sys
+import time
+import uuid
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+# ---------------------------------------------------------------------------
+# Hart verdrahtete autoritative Pfade (NICHT ENV-ueberschreibbar im Evaluator)
+# ---------------------------------------------------------------------------
+_DEFAULT_STATE_DIR = "/opt/control-plane/state"
+_DEFAULT_BOOT_ID_FILE = "/proc/sys/kernel/random/boot_id"
+AUTHORITATIVE_STATE_DIR = _DEFAULT_STATE_DIR
+AUTHORITATIVE_BOOT_ID_FILE = _DEFAULT_BOOT_ID_FILE
+AUDIT_DIR = "/audit"
+
+# ---------------------------------------------------------------------------
+# CP1-Logik-Reuse: SoT-Modul importieren und Pfade hart setzen
+# ---------------------------------------------------------------------------
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import control_reader  # noqa: E402  (SoT-Modul, eine semantische Wahrheit)
+
+# ENV-Ueberschreibbarkeit eliminieren: Pfade hart verdrahten.
+control_reader.STATE_DIR = AUTHORITATIVE_STATE_DIR
+control_reader.BOOT_ID_FILE = AUTHORITATIVE_BOOT_ID_FILE
+
+# ---------------------------------------------------------------------------
+# Statische Action-Class-Whitelist (NUR bekannte non-mutating Decision Classes)
+# ---------------------------------------------------------------------------
+ALLOWED_ACTION_CLASSES = frozenset(
+    {
+        "AUTONOMOUS_TICK_START",
+        "AUTONOMOUS_TICK_CONTINUE",
+    }
+)
+# Noch NICHT erlaubt (spaetere Phasen):
+#   SAVE, DELETE, FORGEJO_WRITE, NOTION_WRITE, TELEGRAM_SEND,
+#   HOST_MUTATION, EXTERNAL_MUTATION
+
+# ---------------------------------------------------------------------------
+# Reason Codes (deterministisch, KEINE freien LLM-Texte als Security Decision)
+# ---------------------------------------------------------------------------
+RC_ALLOW = "ALLOW"
+RC_GLOBAL_AUTONOMY_OFF = "GLOBAL_AUTONOMY_OFF"
+RC_EMERGENCY_ON = "EMERGENCY_ON"
+RC_BOOT_ID_MISMATCH = "BOOT_ID_MISMATCH"
+RC_STATE_MISSING = "STATE_MISSING"
+RC_STATE_MALFORMED = "STATE_MALFORMED"
+RC_BOOT_ID_ERROR = "BOOT_ID_ERROR"
+RC_UNKNOWN_ACTION_CLASS = "UNKNOWN_ACTION_CLASS"
+RC_INVALID_REQUEST = "INVALID_REQUEST"
+RC_INTERNAL_ERROR = "INTERNAL_ERROR"
+
+# Limits
+MAX_REQUEST_BYTES = 4096
+RATE_LIMIT_WINDOW_SEC = 60
+RATE_LIMIT_MAX_REQUESTS = 120  # pro Window (RQ-Zugriff)
+
+# ---------------------------------------------------------------------------
+# Audit (append-only, gegen RQ-Manipulation geschuetzt: eigenes Volume)
+# ---------------------------------------------------------------------------
+def _now_iso():
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _audit(entry: dict):
+    """Schreibt einen Audit-Eintrag append-only. Fail-open fuer Audit (kein Security-Decision)."""
+    try:
+        os.makedirs(AUDIT_DIR, exist_ok=True)
+        path = os.path.join(AUDIT_DIR, "gate_audit.log")
+        line = json.dumps(entry, sort_keys=True) + "\n"
+        with open(path, "a") as f:
+            f.write(line)
+    except Exception:
+        # Audit-Fehler darf die Gate-Entscheidung nicht beeinflussen (fail-open fuer Audit).
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Rate Limiter (einfach, in-memory, pro Client-IP)
+# ---------------------------------------------------------------------------
+class _RateLimiter:
+    def __init__(self, window_sec, max_requests):
+        self.window_sec = window_sec
+        self.max_requests = max_requests
+        self._hits = {}  # ip -> list[timestamp]
+
+    def allow(self, ip: str) -> bool:
+        now = time.time()
+        cutoff = now - self.window_sec
+        hits = [t for t in self._hits.get(ip, []) if t > cutoff]
+        if len(hits) >= self.max_requests:
+            self._hits[ip] = hits
+            return False
+        hits.append(now)
+        self._hits[ip] = hits
+        return True
+
+
+_rate_limiter = _RateLimiter(RATE_LIMIT_WINDOW_SEC, RATE_LIMIT_MAX_REQUESTS)
+
+
+# ---------------------------------------------------------------------------
+# Gate-Entscheidung (deterministisch, fail-closed)
+# ---------------------------------------------------------------------------
+def _effective_state():
+    """Berechnet den Effective-State via CP1-Logik. Fail-closed: Exception -> None."""
+    try:
+        return control_reader.read_control_state()
+    except Exception:
+        return None
+
+
+def _tick_decision(state):
+    """
+    AUTONOMOUS_TICK_START / CONTINUE = ALLOW nur wenn:
+      GLOBAL_AUTONOMY_EFFECTIVE == ON AND EMERGENCY_EFFECTIVE == OFF
+    Alles andere -> DENY mit deterministischem Reason Code.
+    """
+    if state is None:
+        return {"result": "DENY", "reason_code": RC_INTERNAL_ERROR}
+
+    # Emergency: fail-closed, unklar = ON
+    emergency_eff = state.get("emergency_effective")
+    if emergency_eff != "OFF":
+        return {"result": "DENY", "reason_code": RC_EMERGENCY_ON}
+
+    # Boot-ID: muss vorhanden und plausibel sein (explizit, praeziser Reason)
+    current_boot_id = state.get("current_boot_id")
+    if not current_boot_id:
+        return {"result": "DENY", "reason_code": RC_BOOT_ID_ERROR}
+
+    # Global Autonomy: muss ON sein
+    global_eff = state.get("global_autonomy_effective")
+    if global_eff != "ON":
+        return {"result": "DENY", "reason_code": RC_GLOBAL_AUTONOMY_OFF}
+
+    return {"result": "ALLOW", "reason_code": RC_ALLOW}
+
+
+def _check_action(action_class: str):
+    """Validiert Action Class und liefert die Gate-Entscheidung."""
+    if action_class not in ALLOWED_ACTION_CLASSES:
+        return {"result": "DENY", "reason_code": RC_UNKNOWN_ACTION_CLASS}
+
+    state = _effective_state()
+    decision = _tick_decision(state)
+
+    # Audit-Eintrag
+    _audit(
+        {
+            "request_id": str(uuid.uuid4()),
+            "timestamp": _now_iso(),
+            "action_class": action_class,
+            "result": decision["result"],
+            "reason_code": decision["reason_code"],
+            "current_boot_id": (state or {}).get("current_boot_id"),
+            "effective_global": (state or {}).get("global_autonomy_effective"),
+            "effective_emergency": (state or {}).get("emergency_effective"),
+        }
+    )
+    return decision
+
+
+# ---------------------------------------------------------------------------
+# Strict JSON-Schema-Validierung (keine unbekannten Felder, keine Duplikate)
+# ---------------------------------------------------------------------------
+def _validate_check_request(raw_body: bytes):
+    """
+    Validiert den /check-Request streng.
+    Erlaubt NUR: {"action_class": "<string>"}
+    Unbekannte Felder, Duplikate, null, falsche Typen -> INVALID_REQUEST.
+    Rueckgabe: (action_class | None, error_reason | None)
+    """
+    if not raw_body or len(raw_body) > MAX_REQUEST_BYTES:
+        return None, RC_INVALID_REQUEST
+    try:
+        text = raw_body.decode("utf-8")
+    except UnicodeDecodeError:
+        return None, RC_INVALID_REQUEST
+
+    try:
+        obj = json.loads(text)
+    except json.JSONDecodeError:
+        return None, RC_INVALID_REQUEST
+
+    if not isinstance(obj, dict):
+        return None, RC_INVALID_REQUEST
+
+    # Nur das Feld "action_class" erlaubt
+    if set(obj.keys()) != {"action_class"}:
+        return None, RC_INVALID_REQUEST
+
+    action_class = obj.get("action_class")
+    if not isinstance(action_class, str):
+        return None, RC_INVALID_REQUEST
+    if not action_class.strip():
+        return None, RC_INVALID_REQUEST
+
+    return action_class, None
+
+
+# ---------------------------------------------------------------------------
+# HTTP Handler
+# ---------------------------------------------------------------------------
+class _Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+
+    def _send_json(self, status, payload):
+        body = json.dumps(payload, sort_keys=True).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("Cache-Control", "no-store")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _client_ip(self):
+        return self.client_address[0] if self.client_address else "unknown"
+
+    def do_GET(self):
+        ip = self._client_ip()
+        if not _rate_limiter.allow(ip):
+            self._send_json(429, {"error": "rate_limited"})
+            return
+
+        if self.path == "/health":
+            self._send_json(200, {"status": "ok", "service": "gate-evaluator"})
+            return
+        if self.path == "/effective":
+            state = _effective_state()
+            if state is None:
+                self._send_json(500, {"error": "internal_error"})
+                return
+            # Observability-Projektion (KEINE Autoritaet)
+            self._send_json(200, state)
+            return
+        self._send_json(404, {"error": "not_found"})
+
+    def do_POST(self):
+        ip = self._client_ip()
+        if not _rate_limiter.allow(ip):
+            self._send_json(429, {"error": "rate_limited"})
+            return
+
+        if self.path != "/check":
+            self._send_json(404, {"error": "not_found"})
+            return
+
+        # Request-Groesse begrenzen
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            self._send_json(400, {"result": "DENY", "reason_code": RC_INVALID_REQUEST})
+            return
+        if length <= 0 or length > MAX_REQUEST_BYTES:
+            self._send_json(400, {"result": "DENY", "reason_code": RC_INVALID_REQUEST})
+            return
+
+        raw_body = self.rfile.read(length)
+        action_class, err = _validate_check_request(raw_body)
+        if err is not None:
+            self._send_json(400, {"result": "DENY", "reason_code": err})
+            return
+
+        decision = _check_action(action_class)
+        status = 200 if decision["result"] == "ALLOW" else 403
+        self._send_json(status, decision)
+
+    def log_message(self, fmt, *args):
+        # Stille Logs (kein Request-Dump auf stdout)
+        pass
+
+
+def main():
+    port = int(os.environ.get("GATE_EVALUATOR_PORT", "8080"))
+    server = ThreadingHTTPServer(("0.0.0.0", port), _Handler)
+    server.daemon_threads = True
+    server.serve_forever()
+
+
+if __name__ == "__main__":
+    main()

red-queen-architecture/control-plane/cp2a1/test_gate_evaluator.py

@@ -0,0 +1,353 @@
+#!/usr/bin/env python3
+"""
+CP2A1 — Trusted Gate Evaluator Unit-Tests (A-O) + Bypass-Tests (17).
+
+Verwendet Fixture-State (isolierte Test-Verzeichnisse), NICHT den produktiven
+Control-State. Kein produktiver State wird veraendert.
+"""
+
+import json
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import gate_evaluator as ge  # noqa: E402
+
+
+def _write_state(state_dir, name, content):
+    os.makedirs(state_dir, exist_ok=True)
+    with open(os.path.join(state_dir, name), "w") as f:
+        f.write(content)
+
+
+def _make_fixture(global_autonomy="OFF", emergency="OFF", boot_id="boot-1234"):
+    """Baut ein isoliertes Fixture-State-Verzeichnis. Rueckgabe: (state_dir, boot_id)."""
+    tmp = tempfile.mkdtemp(prefix="cp2a1_fixture_")
+    _write_state(tmp, "global_autonomy", global_autonomy)
+    _write_state(tmp, "productive_mutations", "OFF")
+    _write_state(tmp, "save_execution", "OFF")
+    _write_state(tmp, "delete_execution", "OFF")
+    _write_state(tmp, "emergency_stop", emergency)
+    return tmp, boot_id
+
+
+class GateEvaluatorTest(unittest.TestCase):
+    def setUp(self):
+        # Fixture-State isoliert setzen (NICHT produktiv)
+        self._orig_state_dir = ge.AUTHORITATIVE_STATE_DIR
+        self._orig_boot_file = ge.AUTHORITATIVE_BOOT_ID_FILE
+        self._orig_audit_dir = ge.AUDIT_DIR
+        self._fixture_dir, self._boot_id = _make_fixture()
+        ge.AUTHORITATIVE_STATE_DIR = self._fixture_dir
+        ge.AUTHORITATIVE_BOOT_ID_FILE = os.path.join(self._fixture_dir, "boot_id")
+        ge.AUDIT_DIR = os.path.join(self._fixture_dir, "audit")
+        # control_reader auf Fixture zeigen lassen
+        ge.control_reader.STATE_DIR = self._fixture_dir
+        ge.control_reader.BOOT_ID_FILE = ge.AUTHORITATIVE_BOOT_ID_FILE
+        # boot_id-Datei schreiben
+        with open(ge.AUTHORITATIVE_BOOT_ID_FILE, "w") as f:
+            f.write(self._boot_id)
+
+    def tearDown(self):
+        ge.AUTHORITATIVE_STATE_DIR = self._orig_state_dir
+        ge.AUTHORITATIVE_BOOT_ID_FILE = self._orig_boot_file
+        ge.AUDIT_DIR = self._orig_audit_dir
+        ge.control_reader.STATE_DIR = self._orig_state_dir
+        ge.control_reader.BOOT_ID_FILE = self._orig_boot_file
+        shutil.rmtree(self._fixture_dir, ignore_errors=True)
+
+    # --- A: global ON + emergency OFF + valid boot -> ALLOW ---
+    def test_A_global_on_emergency_off_valid_boot_allow(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "ALLOW")
+        self.assertEqual(d["reason_code"], ge.RC_ALLOW)
+
+    # --- B: global OFF -> DENY ---
+    def test_B_global_off_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "OFF")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+    # --- C: emergency ON -> DENY ---
+    def test_C_emergency_on_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "ON")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_EMERGENCY_ON)
+
+    # --- D: grant boot mismatch -> DENY ---
+    def test_D_grant_boot_mismatch_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=other-boot\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+    # --- E: missing global -> DENY ---
+    def test_E_missing_global_deny(self):
+        os.remove(os.path.join(self._fixture_dir, "global_autonomy"))
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+    # --- F: malformed global -> DENY ---
+    def test_F_malformed_global_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "MAYBE")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+    # --- G: missing emergency -> DENY (fail-closed: unklar = ON) ---
+    def test_G_missing_emergency_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        os.remove(os.path.join(self._fixture_dir, "emergency_stop"))
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_EMERGENCY_ON)
+
+    # --- H: malformed emergency -> DENY ---
+    def test_H_malformed_emergency_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "MAYBE")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_EMERGENCY_ON)
+
+    # --- I: missing boot_id -> DENY ---
+    def test_I_missing_boot_id_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        os.remove(ge.AUTHORITATIVE_BOOT_ID_FILE)
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_BOOT_ID_ERROR)
+
+    # --- J: malformed boot_id -> DENY ---
+    def test_J_malformed_boot_id_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        with open(ge.AUTHORITATIVE_BOOT_ID_FILE, "w") as f:
+            f.write("has whitespace\n")
+        d = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_BOOT_ID_ERROR)
+
+    # --- K: unknown action -> DENY ---
+    def test_K_unknown_action_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        for cls in ("SAVE", "DELETE", "FORGEJO_WRITE", "NOTION_WRITE", "TELEGRAM_SEND",
+                    "HOST_MUTATION", "EXTERNAL_MUTATION", "BOGUS"):
+            d = ge._check_action(cls)
+            self.assertEqual(d["result"], "DENY", f"action {cls} should DENY")
+            self.assertEqual(d["reason_code"], ge.RC_UNKNOWN_ACTION_CLASS)
+
+    # --- L: malformed request -> DENY ---
+    def test_L_malformed_request_deny(self):
+        cases = [
+            b"",
+            b"not json",
+            b"[]",
+            b'"string"',
+            b"null",
+            b"{}",
+            b'{"action_class": 123}',
+            b'{"action_class": ""}',
+            b'{"action_class": "AUTONOMOUS_TICK_START", "extra": 1}',
+        ]
+        for body in cases:
+            cls, err = ge._validate_check_request(body)
+            self.assertIsNone(cls, f"body {body!r} should be invalid")
+            self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    # --- M: evaluator exception -> DENY ---
+    def test_M_evaluator_exception_deny(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        # Simuliere Exception in read_control_state -> _effective_state -> None -> DENY
+        orig = ge.control_reader.read_control_state
+        ge.control_reader.read_control_state = lambda: (_ for _ in ()).throw(RuntimeError("boom"))
+        try:
+            d = ge._check_action("AUTONOMOUS_TICK_START")
+        finally:
+            ge.control_reader.read_control_state = orig
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_INTERNAL_ERROR)
+
+    # --- N: cache/stale-state cannot authorize ---
+    def test_N_no_cache_stale_state(self):
+        # Erst ALLOW, dann State auf OFF aendern -> naechste Entscheidung muss DENY sein
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+        d1 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d1["result"], "ALLOW")
+        # State aendern (kein Cache)
+        _write_state(self._fixture_dir, "global_autonomy", "OFF")
+        d2 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d2["result"], "DENY")
+        self.assertEqual(d2["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+    # --- O: RQ-supplied path ignored/rejected ---
+    def test_O_rq_supplied_path_ignored(self):
+        # Der Evaluator akzeptiert KEINEN State-Pfad im Request.
+        # /check-Request mit "state_path" Feld -> INVALID_REQUEST
+        cls, err = ge._validate_check_request(
+            b'{"action_class": "AUTONOMOUS_TICK_START", "state_path": "/etc/passwd"}'
+        )
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+
+class BypassTest(unittest.TestCase):
+    """Fresh isolated checker: versucht Bypass-Pfade. Erwartung: kein Bypass."""
+
+    def setUp(self):
+        self._orig_state_dir = ge.AUTHORITATIVE_STATE_DIR
+        self._orig_boot_file = ge.AUTHORITATIVE_BOOT_ID_FILE
+        self._orig_audit_dir = ge.AUDIT_DIR
+        self._fixture_dir, self._boot_id = _make_fixture()
+        ge.AUTHORITATIVE_STATE_DIR = self._fixture_dir
+        ge.AUTHORITATIVE_BOOT_ID_FILE = os.path.join(self._fixture_dir, "boot_id")
+        ge.AUDIT_DIR = os.path.join(self._fixture_dir, "audit")
+        ge.control_reader.STATE_DIR = self._fixture_dir
+        ge.control_reader.BOOT_ID_FILE = ge.AUTHORITATIVE_BOOT_ID_FILE
+        with open(ge.AUTHORITATIVE_BOOT_ID_FILE, "w") as f:
+            f.write(self._boot_id)
+
+    def tearDown(self):
+        ge.AUTHORITATIVE_STATE_DIR = self._orig_state_dir
+        ge.AUTHORITATIVE_BOOT_ID_FILE = self._orig_boot_file
+        ge.AUDIT_DIR = self._orig_audit_dir
+        ge.control_reader.STATE_DIR = self._orig_state_dir
+        ge.control_reader.BOOT_ID_FILE = self._orig_boot_file
+        shutil.rmtree(self._fixture_dir, ignore_errors=True)
+
+    def _allow_setup(self):
+        _write_state(self._fixture_dir, "global_autonomy", "ON\nboot_id=boot-1234\n")
+        _write_state(self._fixture_dir, "emergency_stop", "OFF")
+
+    def test_alternate_state_path(self):
+        # Request mit state_path -> INVALID_REQUEST (kein Bypass)
+        cls, err = ge._validate_check_request(
+            b'{"action_class": "AUTONOMOUS_TICK_START", "state_path": "/tmp/evil"}'
+        )
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_path_traversal(self):
+        # Path-Traversal-Versuch in action_class -> nicht in Whitelist -> DENY
+        cls, err = ge._validate_check_request(
+            b'{"action_class": "../../etc/passwd"}'
+        )
+        self.assertEqual(cls, "../../etc/passwd")
+        d = ge._check_action(cls)
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_UNKNOWN_ACTION_CLASS)
+
+    def test_env_override(self):
+        # ENV kann die hart verdrahteten Pfade NICHT ueberschreiben.
+        # Die Default-Konstanten (vor Fixture-Ueberschreibung) sind die autoritativen Pfade.
+        self.assertEqual(ge._DEFAULT_STATE_DIR, "/opt/control-plane/state")
+        self.assertEqual(ge._DEFAULT_BOOT_ID_FILE, "/proc/sys/kernel/random/boot_id")
+        # ENV-Variablen werden im Evaluator nicht gelesen (kein os.environ.get fuer Pfade)
+        self.assertNotIn("C5_CONTROL_STATE_DIR", os.environ)
+        self.assertNotIn("C5_BOOT_ID_FILE", os.environ)
+
+    def test_action_injection(self):
+        # Action Class in anderem Feld -> INVALID_REQUEST
+        cls, err = ge._validate_check_request(
+            b'{"action_class": "AUTONOMOUS_TICK_START", "inject": "SAVE"}'
+        )
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_oversized_request(self):
+        big = b'{"action_class": "' + b"A" * 5000 + b'"}'
+        cls, err = ge._validate_check_request(big)
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_duplicate_fields(self):
+        # JSON mit doppeltem action_class: json.loads loest still auf (letzter Wert).
+        # Der Wert wird trotzdem gegen die Whitelist validiert -> kein Bypass.
+        cls, err = ge._validate_check_request(
+            b'{"action_class": "AUTONOMOUS_TICK_START", "action_class": "AUTONOMOUS_TICK_START"}'
+        )
+        self.assertEqual(cls, "AUTONOMOUS_TICK_START")
+        # Duplikat mit bösartigem Wert -> DENY (Whitelist-Check greift)
+        cls2, _ = ge._validate_check_request(
+            b'{"action_class": "AUTONOMOUS_TICK_START", "action_class": "SAVE"}'
+        )
+        self.assertEqual(cls2, "SAVE")
+        d = ge._check_action(cls2)
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_UNKNOWN_ACTION_CLASS)
+
+    def test_null_values(self):
+        cls, err = ge._validate_check_request(b'{"action_class": null}')
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_unicode_confusable_action(self):
+        # Unicode-Konfusables sind nicht in der Whitelist -> DENY
+        cls, err = ge._validate_check_request(b'{"action_class": "AUTONOMOUS_TICK_START\\u200b"}')
+        # Zero-Width-Space -> nicht in Whitelist -> UNKNOWN_ACTION_CLASS
+        self.assertEqual(cls, "AUTONOMOUS_TICK_START\u200b")
+        d = ge._check_action(cls)
+        self.assertEqual(d["result"], "DENY")
+        self.assertEqual(d["reason_code"], ge.RC_UNKNOWN_ACTION_CLASS)
+
+    def test_unknown_json_keys(self):
+        cls, err = ge._validate_check_request(b'{"foo": "bar"}')
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_malformed_json(self):
+        cls, err = ge._validate_check_request(b'{"action_class": "AUTONOMOUS_TICK_START"')
+        self.assertIsNone(cls)
+        self.assertEqual(err, ge.RC_INVALID_REQUEST)
+
+    def test_replay_same_request(self):
+        # Replay desselben Requests: Entscheidung wird frisch berechnet (kein Cache).
+        # Bei State=OFF -> DENY, auch wenn vorher ALLOW war.
+        self._allow_setup()
+        d1 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d1["result"], "ALLOW")
+        _write_state(self._fixture_dir, "global_autonomy", "OFF")
+        d2 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d2["result"], "DENY")
+
+    def test_evaluator_restart(self):
+        # Simuliere Restart: State bleibt, Entscheidung frisch
+        self._allow_setup()
+        d1 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d1["result"], "ALLOW")
+        # "Restart" = neue Instanz (frischer Aufruf)
+        d2 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d2["result"], "ALLOW")
+
+    def test_host_boot_id_fixture_change(self):
+        # boot_id aendern -> Grant boot mismatch -> DENY
+        self._allow_setup()
+        d1 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d1["result"], "ALLOW")
+        with open(ge.AUTHORITATIVE_BOOT_ID_FILE, "w") as f:
+            f.write("new-boot-9999")
+        d2 = ge._check_action("AUTONOMOUS_TICK_START")
+        self.assertEqual(d2["result"], "DENY")
+        self.assertEqual(d2["reason_code"], ge.RC_GLOBAL_AUTONOMY_OFF)
+
+
+if __name__ == "__main__":
+    unittest.main()