diff --git a/red-queen-architecture/control-plane/cp2a1/Dockerfile b/red-queen-architecture/control-plane/cp2a1/Dockerfile new file mode 100644 index 0000000..1adca77 --- /dev/null +++ b/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 +# 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"] diff --git a/red-queen-architecture/control-plane/cp2a1/control_reader.py b/red-queen-architecture/control-plane/cp2a1/control_reader.py new file mode 100755 index 0000000..b9ad721 --- /dev/null +++ b/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="). + - 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() diff --git a/red-queen-architecture/control-plane/cp2a1/docker-compose.yml b/red-queen-architecture/control-plane/cp2a1/docker-compose.yml new file mode 100644 index 0000000..252f847 --- /dev/null +++ b/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: diff --git a/red-queen-architecture/control-plane/cp2a1/gate_evaluator.py b/red-queen-architecture/control-plane/cp2a1/gate_evaluator.py new file mode 100644 index 0000000..96b3b43 --- /dev/null +++ b/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": ""} + 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() diff --git a/red-queen-architecture/control-plane/cp2a1/test_gate_evaluator.py b/red-queen-architecture/control-plane/cp2a1/test_gate_evaluator.py new file mode 100644 index 0000000..0e7a9f1 --- /dev/null +++ b/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()