217 lines
7.5 KiB
Python
217 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
|
"""CP1 statische + adversariale Tests (A-J) für control_reader.py.
|
|
|
|
CP1.0S: trading_execution aus der Hermes Control Plane entfernt (separate
|
|
Security-Domain). Test E (trading-Gate) wurde entfernt, da der Gate nicht mehr
|
|
existiert. Alle übrigen Tests prüfen die 5 Hermes-Controls.
|
|
"""
|
|
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("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 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("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("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("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("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 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")
|
|
# 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("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("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("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("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("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("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("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)
|