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

"""Gegenmessung zu Task 229: Protokollwachstum im gesperrten Zustand, beide Haelften des Schluesselwegs.

Aufruf (WSL, Venv): bremse_echtlauf.py <port> <ordner> <beginn|abschluss>
  beginn    -- fuenf falsche Passwoerter am Schluesselweg, dann 5 s Dauerfeuer auf beginn, dann 5 s auf
               abschluss, dann 5 s auf den Code-Weg. Erwartet: genau EINE Zeile "gesperrt".
  abschluss -- fuenf falsche Anmeldungen am Code-Weg, dann 5 s Dauerfeuer auf abschluss. Erwartet: genau
               eine Zeile "gesperrt" mit weg=schluessel und leerem Benutzer.
Die Probe-Instanz (start_probe_w.py) laeuft aus einer Kopie oder einem Arbeitsbaum ohne Geheimnisdatei, mit
eigener Ablage und eigenem Port. Die Bremse lebt im Prozess: je Durchgang eine frische Instanz starten.
"""

import http.client
import json
import sys
import time
from pathlib import Path

PORT = int(sys.argv[1])
ORDNER = Path(sys.argv[2])
ART = sys.argv[3]
HOST = "localhost:" + str(PORT)
ORIGIN = "http://localhost:" + str(PORT)
PROTOKOLL = ORDNER / "anmeldungen.jsonl"
daten = json.loads((ORDNER / "zugang.json").read_text(encoding="utf-8"))
befunde = 0


def sage(name, ok, text):
    global befunde
    if not ok:
        befunde += 1
    print("[" + ("OK  " if ok else "FAIL") + "] " + name + ": " + str(text), flush=True)


def anfrage(pfad, koerper, host=HOST, origin=ORIGIN):
    verbindung = http.client.HTTPConnection("127.0.0.1", PORT, timeout=30)
    rumpf = json.dumps(koerper).encode("utf-8")
    verbindung.request(
        "POST", pfad, body=rumpf,
        headers={"Host": host, "Origin": origin, "Content-Type": "application/json"},
    )
    antwort = verbindung.getresponse()
    inhalt = antwort.read()
    verbindung.close()
    return antwort.status, inhalt


def zeilen():
    if not PROTOKOLL.exists():
        return []
    return [json.loads(z) for z in PROTOKOLL.read_text(encoding="utf-8").splitlines() if z.strip()]


def groesse():
    return PROTOKOLL.stat().st_size if PROTOKOLL.exists() else 0


def feuer(name, pfad, koerper, sekunden=5.0):
    vorher_b, vorher_z = groesse(), len(zeilen())
    ende = time.monotonic() + sekunden
    stati = {}
    anzahl = 0
    while time.monotonic() < ende:
        status, _ = anfrage(pfad, koerper)
        stati[status] = stati.get(status, 0) + 1
        anzahl += 1
    zuwachs_b, zuwachs_z = groesse() - vorher_b, len(zeilen()) - vorher_z
    print(
        f"     {name}: {anzahl} Anfragen in {sekunden:.1f} s, Stati {stati}, Zuwachs {zuwachs_b} B "
        f"({zuwachs_b / sekunden:.0f} B/s), {zuwachs_z} Zeile(n)",
        flush=True,
    )
    return anzahl, stati, zuwachs_b, zuwachs_z


BEGINN = "/api/anmeldung/schluessel/beginn"
ABSCHLUSS = "/api/anmeldung/schluessel/abschluss"
CODE = "/api/anmeldung"
falsch_beginn = {"benutzer": daten["benutzer"], "passwort": "falsch"}
falsch_code = {"benutzer": daten["benutzer"], "passwort": "falsch", "code": "000000"}
erfunden = {"vorgang": "A" * 43, "antwort": {}}

if ART == "beginn":
    stati = [anfrage(BEGINN, falsch_beginn)[0] for _ in range(5)]
    sage("M1 fuenf falsche Passwoerter am Schluesselweg", stati == [401] * 5, stati)
    n, st, zb, zz = feuer("beginn im gesperrten Zustand", BEGINN, falsch_beginn)
    sage("M2 beginn: alle 429, hoechstens eine Zeile", set(st) == {429} and zz <= 1, f"{zz} Zeile(n), {zb} B")
    n, st, zb, zz = feuer("abschluss im gesperrten Zustand", ABSCHLUSS, erfunden)
    sage("M3 abschluss: alle 429, keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n), {zb} B")
    n, st, zb, zz = feuer("Code-Weg im gesperrten Zustand", CODE, falsch_code)
    sage("M4 Code-Weg: alle 429, keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n), {zb} B")
    alle = zeilen()
    gesperrt = [z for z in alle if z["ergebnis"] == "gesperrt"]
    sage(
        "M5 genau eine Zeile gesperrt, vom Schluesselweg",
        len(gesperrt) == 1 and gesperrt[0].get("weg") == "schluessel" and gesperrt[0]["benutzer"] == daten["benutzer"],
        f"{len(gesperrt)} von {len(alle)} Zeilen: {gesperrt[:2]}",
    )
    status, inhalt = anfrage(ABSCHLUSS, erfunden, host="127.0.0.1:" + str(PORT), origin="http://127.0.0.1:" + str(PORT))
    sage("M6 abschluss unter 127.0.0.1 ist 409 mit der richtigen Adresse", status == 409 and ORIGIN.encode() in inhalt, f"{status} {inhalt[:120]!r}")
else:
    stati = [anfrage(CODE, falsch_code)[0] for _ in range(5)]
    sage("N1 fuenf falsche Anmeldungen am Code-Weg", stati == [401] * 5, stati)
    n, st, zb, zz = feuer("abschluss als erste gesperrte Anfrage", ABSCHLUSS, erfunden)
    sage("N2 abschluss: alle 429, genau eine Zeile", set(st) == {429} and zz == 1, f"{zz} Zeile(n), {zb} B")
    gesperrt = [z for z in zeilen() if z["ergebnis"] == "gesperrt"]
    sage(
        "N3 die eine Zeile traegt weg=schluessel und keinen Benutzer",
        len(gesperrt) == 1 and gesperrt[0].get("weg") == "schluessel" and gesperrt[0]["benutzer"] == "",
        gesperrt[:2],
    )
    n, st, zb, zz = feuer("beginn danach", BEGINN, falsch_beginn, sekunden=2.0)
    sage("N4 beginn danach: keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n)")

text = PROTOKOLL.read_text(encoding="utf-8") if PROTOKOLL.exists() else ""
sage("Z1 kein Passwort im Protokoll", daten["passwort"] not in text and "falsch" not in text.replace('"ergebnis": "falsch"', ""), "geprueft")
print(f"{befunde} Befund(e).")
sys.exit(1 if befunde else 0)