claude-regelwerk/.vault-tools/trading-plattform-pruefung/c_eigenlauf.py

"""Eigener Lauf zu Paket C: Ein entfernter Benutzer verliert Sitzung und offenen Strom, der andere bleibt.

Koordinator-Werkzeug, nicht fuers Repo. Frage (Herkunft: Anforderung Paket C vom 2026-09-18, C1, und die
Plankritik): Endet mit dem Entfernen eines Benutzers aus der Zugangsdatei wirklich alles, was er offen hat --
die HTTP-Sitzung (401), der schreibende Weg zum Killswitch (401) und der schon offene Uebersichtsstrom
(Schliessen mit 1008) --, waehrend ein zweiter Benutzer ungestoert weiterarbeitet? Und antwortet das Backend
auf eine halb geschriebene Zugangsdatei mit einer Ablehnung statt mit einem 500?
Gegenstand ist das Probe-Backend aus start_probe_zugang.py (Attrappe, Wegwerf-Register, Wegwerf-Zugang),
gestartet mit zwei-benutzer. Entfernt wird ueber das Skript des Projekts, nicht von Hand.
Was es nicht sieht: einen Browser (dafuer browserlauf_zugang.py) und den echten Broker.
Regressionsnachweis: --sabotage laesst das Entfernen aus; die Pruefungen auf 401 und 1008 muessen dann rot enden.

Aufruf (WSL, aus dem Codebaum): python c_eigenlauf.py <basis-url> <ablage> [--sabotage]
Exit 0 ohne Befund, 2 mit Befund, 3 wenn eine Voraussetzung fehlt.
"""

import base64
import hashlib
import hmac
import json
import struct
import subprocess
import sys
import time
from pathlib import Path

import requests
from websockets.exceptions import ConnectionClosed
from websockets.sync.client import connect

befunde: list[str] = []


def pruefe(name: str, bedingung: bool, einzelheit: str = "") -> None:
    print(f"  {'ok    ' if bedingung else 'BEFUND'} {name}: {einzelheit}")
    if not bedingung:
        befunde.append(name)


def totp(geheimnis: str, zeit_s: float) -> str:
    schluessel = base64.b32decode(geheimnis.upper() + "=" * (-len(geheimnis) % 8))
    mac = hmac.new(schluessel, struct.pack(">Q", int(zeit_s // 30)), hashlib.sha1).digest()
    versatz = mac[-1] & 0x0F
    zahl = struct.unpack(">I", mac[versatz : versatz + 4])[0] & 0x7FFFFFFF
    return f"{zahl % 1_000_000:06d}"


def anmelden(basis: str, zugang: dict) -> requests.Session:
    sitzung = requests.Session()
    antwort = sitzung.post(
        f"{basis}/api/anmeldung",
        json={"benutzer": zugang["benutzer"], "passwort": zugang["passwort"], "code": totp(zugang["geheimnis"], time.time())},
        headers={"Origin": basis},
        timeout=15,
    )
    pruefe(f"Anmeldung {zugang['benutzer']}", antwort.status_code == 200, f"HTTP {antwort.status_code}")
    return sitzung


def cookie(sitzung: requests.Session) -> str:
    return "; ".join(f"{c.name}={c.value}" for c in sitzung.cookies)


def main() -> int:
    if len(sys.argv) < 3:
        print(__doc__)
        return 3
    basis, ablage = sys.argv[1].rstrip("/"), Path(sys.argv[2])
    sabotage = "--sabotage" in sys.argv[3:]
    if totp("GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ", 59) != "287082":
        print("VORAUSSETZUNG FEHLT: der eigene TOTP-Nachbau trifft den RFC-Vektor nicht")
        return 3
    try:
        erster = json.loads((ablage / "zugang.json").read_text(encoding="utf-8"))
        zweiter = json.loads((ablage / "zugang2.json").read_text(encoding="utf-8"))
    except OSError:
        print("VORAUSSETZUNG FEHLT: zugang.json und zugang2.json in der Ablage (Backend mit zwei-benutzer starten)")
        return 3
    zugang_datei = ablage / "zugang.toml"
    ws_url = basis.replace("http", "ws", 1) + "/ws/uebersicht"

    print("== Beide melden sich an, beide Stroeme laufen")
    s1, s2 = anmelden(basis, erster), anmelden(basis, zweiter)
    with connect(ws_url, additional_headers={"Cookie": cookie(s1)}, origin=basis, open_timeout=5) as ws1, connect(
        ws_url, additional_headers={"Cookie": cookie(s2)}, origin=basis, open_timeout=5
    ) as ws2:
        pruefe("Strom des ersten liefert", "zeit" in str(ws1.recv(timeout=10)), "erster Takt da")
        pruefe("Strom des zweiten liefert", "zeit" in str(ws2.recv(timeout=10)), "erster Takt da")

        print("== Der erste Benutzer wird ueber das Skript entfernt")
        if sabotage:
            print("  sabotage: das Entfernen faellt aus")
        else:
            lauf = subprocess.run(
                [sys.executable, "scripts/zugang.py", "--datei", str(zugang_datei), "entfernen", erster["benutzer"]],
                input=erster["benutzer"] + "\n", capture_output=True, text=True, timeout=60,
            )
            pruefe("Skript entfernt mit Exit 0", lauf.returncode == 0, lauf.stdout.strip().splitlines()[-1][:120] if lauf.stdout.strip() else lauf.stderr[:120])
            liste = subprocess.run(
                [sys.executable, "scripts/zugang.py", "--datei", str(zugang_datei), "liste"], capture_output=True, text=True, timeout=60
            )
            pruefe("liste nennt nur noch den zweiten", zweiter["benutzer"] in liste.stdout and erster["benutzer"] not in liste.stdout, liste.stdout.strip().replace("\n", " | ")[:160])
            pruefe("liste zeigt kein Geheimnis", zweiter["geheimnis"] not in liste.stdout and "scrypt" not in liste.stdout.lower(), f"{len(liste.stdout)} Zeichen")

        print("== Der erste ist draussen")
        antwort = s1.get(f"{basis}/api/uebersicht", timeout=15)
        pruefe("GET /api/uebersicht 401", antwort.status_code == 401, f"HTTP {antwort.status_code}")
        antwort = s1.post(f"{basis}/api/killswitch/anhalten", json={"grund": "probe"}, headers={"Origin": basis}, timeout=15)
        pruefe("POST /api/killswitch/anhalten 401", antwort.status_code == 401, f"HTTP {antwort.status_code}")
        antwort = s1.get(f"{basis}/api/sitzung", timeout=15)
        pruefe("/api/sitzung meldet nicht angemeldet", antwort.status_code == 200 and antwort.json().get("angemeldet") is False, antwort.text[:100])
        beginn = time.monotonic()
        geschlossen, takte = None, 0
        try:
            while time.monotonic() - beginn < 12:
                ws1.recv(timeout=6)
                takte += 1
        except ConnectionClosed as ausnahme:
            geschlossen = ausnahme.rcvd.code if ausnahme.rcvd else None
        except TimeoutError:
            pass
        pruefe("offener Strom des ersten schliesst mit 1008", geschlossen == 1008, f"Code {geschlossen}, davor noch {takte} Takt(e), nach {time.monotonic() - beginn:.1f} s")
        pruefe("hoechstens ein Takt rutscht noch durch", takte <= 1, f"{takte} Takt(e)")

        print("== Der zweite arbeitet weiter")
        antwort = s2.get(f"{basis}/api/uebersicht", timeout=15)
        pruefe("GET /api/uebersicht des zweiten 200", antwort.status_code == 200, f"HTTP {antwort.status_code}")
        try:
            pruefe("Strom des zweiten liefert weiter", "zeit" in str(ws2.recv(timeout=10)), "Takt da")
        except Exception as ausnahme:  # noqa: BLE001 - jeder Abbruch hier ist der Befund
            pruefe("Strom des zweiten liefert weiter", False, repr(ausnahme)[:120])

    print("== Halb geschriebene Zugangsdatei")
    original = zugang_datei.read_bytes()
    try:
        zugang_datei.write_text('[benutzer.halb]\npasswort = "x"\n', encoding="utf-8")
        time.sleep(0.05)
        antwort = s2.get(f"{basis}/api/uebersicht", timeout=15)
        pruefe("Ablehnung statt 500", antwort.status_code in (401, 503), f"HTTP {antwort.status_code}")
    finally:
        zugang_datei.write_bytes(original)
    time.sleep(0.05)
    antwort = s2.get(f"{basis}/api/uebersicht", timeout=15)
    # Seit der Nachbesserung vom 2026-09-18: Eine unlesbare Datei sperrt nur aus, solange sie unlesbar ist.
    pruefe("nach dem Zuruecklegen gilt die Sitzung des zweiten wieder", antwort.status_code == 200, f"HTTP {antwort.status_code}")

    print(f"{len(befunde)} Befund(e).")
    return 2 if befunde else 0


if __name__ == "__main__":
    sys.exit(main())