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

"""Angriffsfaelle fuer die Quellen wsl und dockerhost von abendkontrolle.py (Paket DH), unter Windows, ohne Netz.

Koordinator-Werkzeug, nicht fuers Repo. Ersetzt ssh.exe durch eine .cmd-Attrappe, die ein kleines
Python-Skript startet; der Vault ist ein Wegwerf-Ordner. Die Attrappe schreibt jeden Aufruf mit
und antwortet je nach ANGRIFF_ART mit einem Ergebnis des angefragten Teils (JJJJ-MM-TT oder
JJJJ-MM-TT-ibkr), mit Exit 3 (noch nichts da), mit Muell oder mit einem Ergebnis des falschen
Teils. Der Treiber faengt je Fall auch SystemExit: Gegen ein abendkontrolle.py ohne --quelle
bricht argparse damit ab, und das soll ein Fall sein, nicht das Ende des Laufs.
Abgedichtet (Plankritik DH, 2026-09-26): WSL zeigt in jedem Lauf ins Leere, und fahre_wsl ist
ohne eigene Attrappe des Falls eine Attrappe, die ablehnt; ein Aufruf des echten Wegs ist eine
Abweichung, kein Lauf. Zwei Faelle belegen beide Sperren.
Aufruf: python abendkontrolle_dockerhost_angriffe.py
Exit 0 ohne Abweichung, 2 mit Abweichung.
"""

import contextlib
import importlib.util
import io
import json
import os
import sys
import tempfile
from datetime import datetime
from pathlib import Path

HIER = Path(__file__).resolve().parent

ATTRAPPE_PY = r'''
import json
import os
import sys

with open(os.environ["ANGRIFF_PROTOKOLL"], "a", encoding="utf-8") as datei:
    datei.write(json.dumps(sys.argv[1:]) + "\n")
art = os.environ.get("ANGRIFF_ART", "gut")
if art == "fehlt":
    sys.stderr.write("abendkontrolle-lesen: fuer diesen Tag liegt noch kein Ergebnis\n")
    sys.exit(3)
if art == "muell":
    print("kein json")
    sys.exit(0)
ziel = sys.argv[-1]
ibkr = ziel.endswith("-ibkr")
tag = ziel[:-5] if ibkr else ziel
teil = "ibkr" if ibkr and art != "falscher_teil" else "alpaca"
print(json.dumps({"tag": tag, "teil": teil, "befunde": [], "hinweise": [], "fakten": {"handelstag": True}, "exit": 0}))
'''

FAELLE = []


def fall(funktion):
    FAELLE.append(funktion)
    return funktion


def lade():
    spec = importlib.util.spec_from_file_location("abendkontrolle_unter_angriff", HIER / "abendkontrolle.py")
    modul = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(modul)
    return modul


def lauf(art, *argumente, modul=None):
    """Ein Lauf von main() in einem Wegwerf-Vault; liefert Exit, Notiz, Ergebnisdateien und ssh-Aufrufe."""
    with tempfile.TemporaryDirectory() as ordner:
        wurzel = Path(ordner)
        vault = wurzel / "vault"
        (vault / "Inbox").mkdir(parents=True)
        protokoll = wurzel / "ssh.jsonl"
        attrappe = wurzel / "ssh_attrappe.py"
        attrappe.write_text(ATTRAPPE_PY, encoding="utf-8")
        cmd = wurzel / "ssh_attrappe.cmd"
        cmd.write_text(f'@"{sys.executable}" "{attrappe}" %*\r\n', encoding="utf-8")
        modul = modul or lade()
        modul.SSH = str(cmd)
        # Abdichtung (Plankritik DH, 2026-09-26): Der echte WSL-Weg lief sonst gegen den echten
        # Projektordner, fragte das Gateway und schickte echte ntfy-Meldungen. wsl.exe und das
        # Skript zeigen ins Leere, und fahre_wsl ist eine Attrappe, die jeden Aufruf samt beiden
        # Pfaden mitschreibt und ablehnt -- ausser ein Fall hat seine eigene gesetzt (wsl_attrappe).
        modul.WSL = str(wurzel / "kein-wsl" / "wsl.exe")
        modul.WSL_SKRIPT = "/nicht-vorhanden/abendkontrolle_wsl.py"
        unerwartet = []
        if not getattr(modul.fahre_wsl, "attrappe", False):

            def wsl_verboten(tag, teil="alpaca"):
                unerwartet.append((tag, teil, modul.WSL, modul.WSL_SKRIPT))
                raise RuntimeError("echter WSL-Weg ohne Attrappe")

            modul.fahre_wsl = wsl_verboten
        os.environ["ANGRIFF_PROTOKOLL"] = str(protokoll)
        os.environ["ANGRIFF_ART"] = art
        code = modul.main(["--vault", str(vault), *argumente])
        if unerwartet:
            raise AssertionError(f"WSL-Weg ohne Attrappe gerufen: {unerwartet}")
        notizen = sorted((vault / "Inbox").glob("*.md"))
        notiz = "\n".join(n.read_text(encoding="utf-8") for n in notizen)
        ablage = vault / modul.ABLAGE_TEIL
        ergebnisse = sorted(p.name for p in ablage.glob("*.json")) if ablage.exists() else []
        aufrufe = []
        if protokoll.exists():
            aufrufe = [json.loads(z) for z in protokoll.read_text(encoding="utf-8").splitlines() if z.strip()]
        return code, notiz, ergebnisse, aufrufe


def wsl_attrappe(modul, gerufen):
    """Ersetzt fahre_wsl im Modul; schreibt (tag, teil) mit und liefert ein Ergebnis des Teils."""

    def fahre_wsl(tag, teil="alpaca"):
        gerufen.append((tag, teil))
        return {"tag": tag, "teil": teil, "befunde": [], "hinweise": [], "fakten": {}, "exit": 0}

    fahre_wsl.attrappe = True
    modul.fahre_wsl = fahre_wsl


@fall
def ergebnis_vom_host_landet_in_ablage_und_notiz():
    code, notiz, ergebnisse, aufrufe = lauf(
        "gut", "--quelle", "dockerhost", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30"
    )
    assert code == 0, code
    assert ergebnisse == ["2026-10-05-ibkr.json", "2026-10-05.json"], ergebnisse
    assert "Abendkontrolle Trading Plattform (automatisch): ohne Befund" in notiz, notiz
    assert "Abendkontrolle IBKR (automatisch): ohne Befund" in notiz, notiz
    assert len(aufrufe) == 2, aufrufe


@fall
def ssh_mit_festem_schluessel_ohne_rueckfrage_nur_zum_bekannten_host():
    _, _, _, aufrufe = lauf("gut", "--quelle", "dockerhost", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30")
    for argumente, ziel in zip(aufrufe, ("2026-10-05", "2026-10-05-ibkr"), strict=True):
        assert argumente[:3] == ["-p", "2498", "-i"], argumente
        assert argumente[3].endswith("trading-dh-kontrolle"), argumente
        assert argumente[4:14] == [
            "-o", "BatchMode=yes",
            "-o", "IdentitiesOnly=yes",
            "-o", "StrictHostKeyChecking=yes",
            "-o", "UserKnownHostsFile=~/.ssh/known_hosts_trading_dh",
            "-o", "ConnectTimeout=20",
        ], argumente
        assert argumente[14:] == ["[email protected]", ziel], argumente


@fall
def noch_kein_ergebnis_ist_ein_befund_in_der_notiz_und_wird_wiederholt():
    code, notiz, ergebnisse, _ = lauf(
        "fehlt", "--quelle", "dockerhost", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30"
    )
    assert code == 1, code
    assert ergebnisse == [], ergebnisse
    assert "**gescheitert**" in notiz and "noch kein Ergebnis" in notiz, notiz
    assert "#needs-review" in notiz, notiz
    assert "fuer 2026-10-05-ibkr noch kein Ergebnis" in notiz, notiz


@fall
def muell_vom_host_ist_ein_befund():
    code, notiz, ergebnisse, _ = lauf(
        "muell", "--quelle", "dockerhost", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30"
    )
    assert code == 1, code
    assert ergebnisse == [], ergebnisse
    assert "keine JSON-Zeile vom Docker-Host" in notiz, notiz


@fall
def faellige_tage_kommen_einzeln_ueber_ssh():
    modul = lade()
    _, _, ergebnisse, aufrufe = lauf("gut", "--quelle", "dockerhost", "--jetzt", "2026-10-05T22:10", modul=modul)
    tage = ["2026-09-28", "2026-09-29", "2026-09-30", "2026-10-01", "2026-10-02"]
    ibkr = [f"{t}-ibkr" for t in tage if t >= modul.IBKR_ERSTER_TAG.isoformat()]
    ziele = [a[-1] for a in aufrufe]
    assert ziele == tage + ibkr, ziele
    assert ergebnisse == sorted(f"{z}.json" for z in ziele), ergebnisse


@fall
def ohne_schalter_bleibt_es_bei_wsl():
    modul = lade()
    gerufen = []
    wsl_attrappe(modul, gerufen)

    def verboten(*_argumente):
        raise AssertionError("ohne --quelle darf nichts ueber SSH laufen")

    modul.fahre_dockerhost = verboten
    code, _, ergebnisse, aufrufe = lauf("gut", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30", modul=modul)
    assert code == 0, code
    assert gerufen == [("2026-10-05", "alpaca"), ("2026-10-05", "ibkr")], gerufen
    assert aufrufe == [], aufrufe
    assert ergebnisse == ["2026-10-05-ibkr.json", "2026-10-05.json"], ergebnisse


@fall
def die_ibkr_instanz_bleibt_mit_quelle_ibkr_wsl_in_wsl():
    modul = lade()
    gerufen = []
    wsl_attrappe(modul, gerufen)
    code, _, ergebnisse, aufrufe = lauf(
        "gut", "--quelle", "dockerhost", "--quelle-ibkr", "wsl", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30",
        modul=modul,
    )
    assert code == 0, code
    assert [a[-1] for a in aufrufe] == ["2026-10-05"], aufrufe
    assert gerufen == [("2026-10-05", "ibkr")], gerufen
    assert ergebnisse == ["2026-10-05-ibkr.json", "2026-10-05.json"], ergebnisse


@fall
def ein_ergebnis_des_falschen_teils_vom_host_gilt_als_gescheitert():
    code, notiz, ergebnisse, aufrufe = lauf(
        "falscher_teil", "--quelle", "dockerhost", "--teil", "ibkr", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30"
    )
    assert code == 1, code
    assert [a[-1] for a in aufrufe] == ["2026-10-05-ibkr"], aufrufe
    assert ergebnisse == [], ergebnisse
    assert "gehoert zu 'alpaca'" in notiz and "**gescheitert**" in notiz, notiz


@fall
def eine_unbekannte_quelle_ist_ein_bedienfehler():
    fehler = io.StringIO()
    try:
        with contextlib.redirect_stderr(fehler):
            lauf("gut", "--quelle", "irgendwo", "--tag", "2026-10-05")
    except SystemExit as ende:
        assert ende.code == 2, ende.code
        assert "--quelle: invalid choice" in fehler.getvalue(), fehler.getvalue()
        return
    raise AssertionError("argparse hat die Quelle irgendwo angenommen")


@fall
def die_heutige_frist_und_die_aufgabe_passen_zueinander():
    quelle = (HIER / "abendkontrolle.py").read_text(encoding="utf-8")
    aufgabe = (HIER / "abendkontrolle_einrichten.ps1").read_text(encoding="utf-8")
    assert "StrictHostKeyChecking=no" not in quelle
    assert "if tag == jetzt.date() and jetzt.time() < FRUEHESTENS:" in quelle
    assert aufgabe.count('[ValidateSet("wsl", "dockerhost")]') == 2
    assert "'\" --quelle ' + $Quelle + ' --quelle-ibkr ' + $QuelleIbkr" in aufgabe
    assert 'if ($Quelle -eq "dockerhost" -or $QuelleIbkr -eq "dockerhost") { $Uhrzeit = "23:15" }' in aufgabe
    assert "if (-not $QuelleIbkr) { $QuelleIbkr = $Quelle }" in aufgabe


@fall
def der_echte_wsl_weg_ist_versperrt():
    # Ohne --quelle und ohne eigene Attrappe nimmt main den WSL-Weg. Er muss als Abweichung
    # enden, nicht als Lauf.
    try:
        lauf("gut", "--tag", "2026-10-05", "--jetzt", "2026-10-05T23:30")
    except AssertionError as ausnahme:
        text = str(ausnahme)
        assert "WSL-Weg ohne Attrappe" in text, text
        # Beleg fuer die Zeilen in lauf(): Beim Aufruf zeigten wsl.exe und das Skript ins Leere.
        assert "kein-wsl" in text and "/nicht-vorhanden/" in text, text
        return
    raise AssertionError("ein Lauf ohne WSL-Attrappe kam durch")


@fall
def der_echte_wsl_aufruf_findet_kein_wsl():
    # Die zweite Sperre: Selbst die echte fahre_wsl findet in der Abdichtung kein wsl.exe, und das
    # Skript, das sie dort starten wuerde, gibt es nicht.
    modul = lade()
    with tempfile.TemporaryDirectory() as ordner:
        modul.WSL = str(Path(ordner) / "kein-wsl" / "wsl.exe")
        modul.WSL_SKRIPT = "/nicht-vorhanden/abendkontrolle_wsl.py"
        try:
            modul.fahre_wsl("2026-10-05", "alpaca")
        except FileNotFoundError:
            return
    raise AssertionError("fahre_wsl lief ohne wsl.exe weiter")


def main() -> int:
    abweichungen = 0
    for funktion in FAELLE:
        try:
            funktion()
        except (Exception, SystemExit) as ausnahme:
            abweichungen += 1
            print(f"ABWEICHUNG {funktion.__name__}: {type(ausnahme).__name__}: {ausnahme}")
        else:
            print(f"OK {funktion.__name__}")
    print(f"{len(FAELLE)} Faelle, {abweichungen} Abweichung(en) ({datetime.now():%Y-%m-%d %H:%M})")
    return 2 if abweichungen else 0


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