Automatischer Push: 8 Datei(en) geändert
- .vault-tools/trading-plattform-pruefung/LIESMICH.md - .vault-tools/trading-plattform-pruefung/angriffe.py - .vault-tools/trading-plattform-pruefung/browserlauf_zugang.py - .vault-tools/trading-plattform-pruefung/handabruf_kerzen.py - .vault-tools/trading-plattform-pruefung/lies_paper_protokoll.py - .vault-tools/trading-plattform-pruefung/pruef_beobachtung_echt.py - .vault-tools/trading-plattform-pruefung/start_probe_backend.py - .vault-tools/trading-plattform-pruefung/start_probe_zugang.py Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py. Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5 committed
1 parent
3bb28e4b3c
commit
371a1f5d2a
8 files changed
+890
-3
No files matched your search
@@ -0,0 +1,297 @@
|
||||
"""Angriffsfaelle gegen ein laufendes Dashboard-Backend mit Anmeldung (Teil 10, Teil B).
|
||||
|
||||
Koordinator-Pruefskript, nicht fuers Repo. Es beurteilt, ob ein LAUFENDES Backend die Zusagen der
|
||||
Spec zu Teil 10 haelt; es ist ein Fangnetz, kein Nachweis.
|
||||
|
||||
Geltungsbereich (Herkunft jeder Erwartung: Projektvorgabe, Spec Teil 10, Abschnitt "Teil B"):
|
||||
- freie Pfade (/api/gesund, GET /api/sitzung), geschuetzte Pfade ohne Cookie 401, fremder Host 400,
|
||||
schreibende Anfrage ohne oder mit fremdem Origin 403 (auch mit gueltigem Cookie), /docs /redoc
|
||||
/openapi.json 404, Sicherheitskopfzeilen, einheitliche Ablehnung der Anmeldung, Cookie-Merkmale,
|
||||
Wiederholung eines TOTP-Codes, Abmeldung entwertet das Cookie, WebSocket ohne Cookie und mit
|
||||
fremdem Origin abgelehnt, Bremse (429 mit Retry-After, auch fuer die richtige Anmeldung),
|
||||
Anmeldeprotokoll ohne Passwort und Code, /api/paper mit Protokollpfad ausserhalb des Ordners.
|
||||
Was es nicht sieht: TLS und das Secure-Merkmal (das Probe-Backend spricht HTTP), DNS-Rebinding im
|
||||
echten Browser, Zeitverhalten der Passwortpruefung, alles hinter der Anmeldung ausser einem GET und
|
||||
einem POST je Fall, und ob die CSP die gebaute Oberflaeche wirklich traegt (dafuer der Browser-Lauf).
|
||||
Regressionsnachweis: gegen das alte Probe-Backend ohne Anmeldung (start_probe_backend.py) muss der
|
||||
Lauf mit Exit 2 enden und die Schutzfaelle einzeln rot zeigen.
|
||||
|
||||
Aufruf (in WSL, Venv des Projekts):
|
||||
python angriffe.py <basis-url> <ablageordner des probe-backends> [--ohne-bremse]
|
||||
Exit 0 ohne Befund, 2 mit Befund, 3 wenn eine Voraussetzung fehlt (auch das ist ein Befund).
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import struct
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import requests
|
||||
from websockets.exceptions import InvalidStatus
|
||||
from websockets.sync.client import connect
|
||||
|
||||
befunde: list[str] = []
|
||||
bestanden: list[str] = []
|
||||
|
||||
|
||||
def pruefe(name: str, bedingung: bool, einzelheit: str = "") -> None:
|
||||
if bedingung:
|
||||
bestanden.append(name)
|
||||
print(f" ok {name}")
|
||||
else:
|
||||
befunde.append(f"{name}: {einzelheit}")
|
||||
print(f" ROT {name}: {einzelheit}")
|
||||
|
||||
|
||||
def totp(geheimnis: str, zeit_s: float) -> str:
|
||||
"""Eigener Nachbau nach RFC 6238 / RFC 4226, unabhaengig vom Projektcode."""
|
||||
schluessel = base64.b32decode(geheimnis.upper() + "=" * (-len(geheimnis) % 8))
|
||||
zaehler = int(zeit_s // 30)
|
||||
mac = hmac.new(schluessel, struct.pack(">Q", zaehler), 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 main() -> int:
|
||||
if len(sys.argv) < 3:
|
||||
print(__doc__)
|
||||
return 3
|
||||
basis = sys.argv[1].rstrip("/")
|
||||
ablage = Path(sys.argv[2])
|
||||
mit_bremse = "--ohne-bremse" not in sys.argv[3:]
|
||||
host = basis.split("://", 1)[1]
|
||||
origin = basis
|
||||
|
||||
zugang_json = ablage / "zugang.json"
|
||||
if not zugang_json.is_file():
|
||||
print(f"VORAUSSETZUNG FEHLT: {zugang_json} (Ablage des Probe-Backends)")
|
||||
return 3
|
||||
zugang = json.loads(zugang_json.read_text(encoding="utf-8"))
|
||||
benutzer, passwort, geheimnis = zugang["benutzer"], zugang["passwort"], zugang["geheimnis"]
|
||||
|
||||
# RFC-6238-Vektor gegen den eigenen Nachbau, bevor irgendetwas damit gemessen wird.
|
||||
if totp("GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ", 59) != "287082":
|
||||
print("VORAUSSETZUNG FEHLT: der eigene TOTP-Nachbau trifft den RFC-Vektor nicht")
|
||||
return 3
|
||||
|
||||
try:
|
||||
grund = requests.get(f"{basis}/api/gesund", timeout=5)
|
||||
except requests.RequestException as ausnahme:
|
||||
print(f"VORAUSSETZUNG FEHLT: Backend unter {basis} nicht erreichbar: {ausnahme!r}")
|
||||
return 3
|
||||
|
||||
print("== Grundlinie")
|
||||
pruefe("gesund ohne Cookie 200", grund.status_code == 200, f"HTTP {grund.status_code}")
|
||||
sitzung = requests.get(f"{basis}/api/sitzung", timeout=5)
|
||||
try:
|
||||
sitzung_json = sitzung.json()
|
||||
except ValueError:
|
||||
sitzung_json = {}
|
||||
pruefe(
|
||||
"sitzung ohne Cookie: pflicht, eingerichtet, nicht angemeldet",
|
||||
sitzung.status_code == 200
|
||||
and sitzung_json.get("pflicht") is True
|
||||
and sitzung_json.get("eingerichtet") is True
|
||||
and sitzung_json.get("angemeldet") is False,
|
||||
f"HTTP {sitzung.status_code} {sitzung.text[:120]}",
|
||||
)
|
||||
|
||||
print("== Geschuetzte Pfade ohne Cookie")
|
||||
for pfad in (
|
||||
"/api/uebersicht",
|
||||
"/api/agenten",
|
||||
"/api/bestand",
|
||||
"/api/laeufe",
|
||||
"/api/markt/datensaetze",
|
||||
"/api/paper",
|
||||
):
|
||||
antwort = requests.get(f"{basis}{pfad}", timeout=10)
|
||||
pruefe(f"GET {pfad} ohne Cookie 401", antwort.status_code == 401, f"HTTP {antwort.status_code}")
|
||||
|
||||
print("== Fremder Host")
|
||||
for pfad in ("/", "/api/gesund", "/api/uebersicht"):
|
||||
antwort = requests.get(f"{basis}{pfad}", headers={"Host": "boese.example"}, timeout=5)
|
||||
pruefe(f"GET {pfad} mit fremdem Host 400", antwort.status_code == 400, f"HTTP {antwort.status_code}")
|
||||
|
||||
print("== Schnittstellenbeschreibung")
|
||||
for pfad in ("/docs", "/redoc", "/openapi.json"):
|
||||
antwort = requests.get(f"{basis}{pfad}", timeout=5)
|
||||
pruefe(f"GET {pfad} 404", antwort.status_code == 404, f"HTTP {antwort.status_code}")
|
||||
|
||||
print("== Kopfzeilen")
|
||||
for pfad in ("/", "/api/gesund"):
|
||||
kopf = requests.get(f"{basis}{pfad}", timeout=5).headers
|
||||
csp = kopf.get("content-security-policy", "")
|
||||
pruefe(f"{pfad}: CSP mit script-src 'self'", "script-src 'self'" in csp, csp[:80])
|
||||
pruefe(f"{pfad}: CSP mit frame-ancestors 'none'", "frame-ancestors 'none'" in csp, csp[:80])
|
||||
pruefe(f"{pfad}: X-Frame-Options DENY", kopf.get("x-frame-options") == "DENY", str(kopf.get("x-frame-options")))
|
||||
pruefe(f"{pfad}: nosniff", kopf.get("x-content-type-options") == "nosniff", str(kopf.get("x-content-type-options")))
|
||||
pruefe(f"{pfad}: Referrer-Policy", kopf.get("referrer-policy") == "no-referrer", str(kopf.get("referrer-policy")))
|
||||
pruefe(f"{pfad}: kein Server-Kopf", "server" not in kopf, str(kopf.get("server")))
|
||||
api_kopf = requests.get(f"{basis}/api/gesund", timeout=5).headers
|
||||
pruefe("/api/gesund: Cache-Control no-store", api_kopf.get("cache-control") == "no-store", str(api_kopf.get("cache-control")))
|
||||
|
||||
print("== Anmeldung: Origin und einheitliche Ablehnung")
|
||||
rumpf = {"benutzer": benutzer, "passwort": passwort, "code": totp(geheimnis, time.time())}
|
||||
ohne_origin = requests.post(f"{basis}/api/anmeldung", json=rumpf, timeout=10)
|
||||
pruefe("Anmeldung ohne Origin 403", ohne_origin.status_code == 403, f"HTTP {ohne_origin.status_code}")
|
||||
fremd = requests.post(
|
||||
f"{basis}/api/anmeldung", json=rumpf, headers={"Origin": "http://boese.example"}, timeout=10
|
||||
)
|
||||
pruefe("Anmeldung mit fremdem Origin 403", fremd.status_code == 403, f"HTTP {fremd.status_code}")
|
||||
|
||||
kopf = {"Origin": origin}
|
||||
texte = set()
|
||||
for name, daten in (
|
||||
("falsches Passwort", {**rumpf, "passwort": "ganz-falsches-passwort"}),
|
||||
("falscher Code", {**rumpf, "code": "000000" if rumpf["code"] != "000000" else "111111"}),
|
||||
("unbekannter Benutzer", {**rumpf, "benutzer": "niemand"}),
|
||||
):
|
||||
antwort = requests.post(f"{basis}/api/anmeldung", json=daten, headers=kopf, timeout=10)
|
||||
pruefe(f"Anmeldung, {name}: 401", antwort.status_code == 401, f"HTTP {antwort.status_code}")
|
||||
texte.add(antwort.text)
|
||||
pruefe("alle drei Ablehnungen lauten gleich", len(texte) == 1, str(texte)[:200])
|
||||
|
||||
print("== Anmeldung: Erfolg, Cookie, Wiederholung")
|
||||
jetzt = time.time()
|
||||
code = totp(geheimnis, jetzt)
|
||||
sitz = requests.Session()
|
||||
gut = sitz.post(f"{basis}/api/anmeldung", json={**rumpf, "code": code}, headers=kopf, timeout=10)
|
||||
pruefe("richtige Anmeldung 200", gut.status_code == 200, f"HTTP {gut.status_code} {gut.text[:120]}")
|
||||
setzen = gut.headers.get("set-cookie", "")
|
||||
klein = setzen.lower()
|
||||
pruefe("Cookie tp_sitzung gesetzt", setzen.startswith("tp_sitzung="), setzen[:40])
|
||||
pruefe("Cookie HttpOnly", "httponly" in klein, setzen[-80:])
|
||||
pruefe("Cookie SameSite=Strict", "samesite=strict" in klein, setzen[-80:])
|
||||
pruefe("Cookie Path=/", "path=/" in klein, setzen[-80:])
|
||||
pruefe("Cookie ohne Secure (HTTP)", "secure" not in klein, setzen[-80:])
|
||||
|
||||
wieder = requests.post(f"{basis}/api/anmeldung", json={**rumpf, "code": code}, headers=kopf, timeout=10)
|
||||
pruefe("derselbe Code ein zweites Mal 401", wieder.status_code == 401, f"HTTP {wieder.status_code}")
|
||||
|
||||
print("== Mit Cookie")
|
||||
ueber = sitz.get(f"{basis}/api/uebersicht", timeout=15)
|
||||
pruefe("GET /api/uebersicht mit Cookie 200", ueber.status_code == 200, f"HTTP {ueber.status_code}")
|
||||
halt = sitz.post(f"{basis}/api/killswitch/anhalten", json={"grund": "angriffe.py"}, timeout=15)
|
||||
pruefe("POST mit Cookie, ohne Origin 403", halt.status_code == 403, f"HTTP {halt.status_code}")
|
||||
halt = sitz.post(
|
||||
f"{basis}/api/killswitch/anhalten",
|
||||
json={"grund": "angriffe.py"},
|
||||
headers={"Origin": "http://boese.example"},
|
||||
timeout=15,
|
||||
)
|
||||
pruefe("POST mit Cookie, fremder Origin 403", halt.status_code == 403, f"HTTP {halt.status_code}")
|
||||
halt = sitz.post(
|
||||
f"{basis}/api/killswitch/anhalten", json={"grund": "angriffe.py"}, headers=kopf, timeout=15
|
||||
)
|
||||
pruefe(
|
||||
"POST mit Cookie und Origin kommt an (200 oder 409)",
|
||||
halt.status_code in (200, 409),
|
||||
f"HTTP {halt.status_code} {halt.text[:120]}",
|
||||
)
|
||||
|
||||
print("== /api/paper")
|
||||
paper = sitz.get(f"{basis}/api/paper", timeout=10)
|
||||
pruefe("GET /api/paper mit Cookie 200", paper.status_code == 200, f"HTTP {paper.status_code}")
|
||||
status_datei = ablage / "status.json"
|
||||
if status_datei.is_file() and paper.status_code == 200:
|
||||
daten = paper.json()
|
||||
pruefe("paper: Status und Zeilen geliefert", bool(daten.get("status")) and len(daten.get("zeilen", [])) > 0, str(daten)[:160])
|
||||
original = status_datei.read_text(encoding="utf-8")
|
||||
verbogen = json.loads(original)
|
||||
verbogen["protokoll"] = str(ablage / "registry.db")
|
||||
status_datei.write_text(json.dumps(verbogen), encoding="utf-8")
|
||||
try:
|
||||
aussen = sitz.get(f"{basis}/api/paper", timeout=10).json()
|
||||
pruefe(
|
||||
"paper: Protokollpfad ausserhalb des Ordners liefert keine Zeilen, aber einen Hinweis",
|
||||
aussen.get("zeilen") == [] and bool(aussen.get("hinweis")),
|
||||
str(aussen)[:200],
|
||||
)
|
||||
finally:
|
||||
status_datei.write_text(original, encoding="utf-8")
|
||||
else:
|
||||
print(" (kein Paper-Status in der Ablage; Fall uebersprungen — Backend mit 'mit-paperstatus' starten)")
|
||||
|
||||
print("== WebSocket")
|
||||
ws_url = basis.replace("http", "ws", 1) + "/ws/uebersicht"
|
||||
cookie = "; ".join(f"{c.name}={c.value}" for c in sitz.cookies)
|
||||
|
||||
def ws_versuch(kopfzeilen: dict[str, str], mit_origin: str | None) -> str:
|
||||
try:
|
||||
with connect(ws_url, additional_headers=kopfzeilen, origin=mit_origin, open_timeout=5) as ws:
|
||||
return "offen:" + str(ws.recv(timeout=10))[:40]
|
||||
except InvalidStatus as ausnahme:
|
||||
return f"abgelehnt:{ausnahme.response.status_code}"
|
||||
except Exception as ausnahme: # noqa: BLE001 - jede andere Ablehnung ist auch eine
|
||||
return f"abgelehnt:{ausnahme!r}"[:80]
|
||||
|
||||
pruefe("WebSocket ohne Cookie abgelehnt", ws_versuch({}, origin).startswith("abgelehnt"), ws_versuch({}, origin))
|
||||
ergebnis = ws_versuch({"Cookie": cookie}, "http://boese.example")
|
||||
pruefe("WebSocket mit Cookie, fremder Origin abgelehnt", ergebnis.startswith("abgelehnt"), ergebnis)
|
||||
ergebnis = ws_versuch({"Cookie": cookie}, origin)
|
||||
pruefe("WebSocket mit Cookie und Origin liefert einen Snapshot", ergebnis.startswith("offen:"), ergebnis)
|
||||
|
||||
print("== Abmeldung")
|
||||
ab = sitz.post(f"{basis}/api/abmeldung", headers=kopf, timeout=10)
|
||||
pruefe("Abmeldung 204", ab.status_code == 204, f"HTTP {ab.status_code}")
|
||||
danach = requests.get(f"{basis}/api/uebersicht", headers={"Cookie": cookie}, timeout=10)
|
||||
pruefe("altes Cookie nach der Abmeldung 401", danach.status_code == 401, f"HTTP {danach.status_code}")
|
||||
|
||||
if mit_bremse:
|
||||
print("== Bremse (sperrt diese Adresse; deshalb zuletzt)")
|
||||
status = []
|
||||
for _ in range(12):
|
||||
antwort = requests.post(
|
||||
f"{basis}/api/anmeldung",
|
||||
json={**rumpf, "passwort": "ganz-falsches-passwort"},
|
||||
headers=kopf,
|
||||
timeout=10,
|
||||
)
|
||||
status.append(antwort.status_code)
|
||||
if antwort.status_code == 429:
|
||||
pruefe("429 traegt Retry-After", antwort.headers.get("retry-after", "").isdigit(), str(antwort.headers.get("retry-after")))
|
||||
break
|
||||
pruefe("Bremse greift innerhalb von zwoelf Fehlversuchen", 429 in status, str(status))
|
||||
# Ein frischer Zeitschritt, damit die Ablehnung nicht an der Wiederholung des Codes liegt.
|
||||
warten = 30 - (time.time() % 30) + 1
|
||||
time.sleep(warten)
|
||||
richtig = requests.post(
|
||||
f"{basis}/api/anmeldung",
|
||||
json={**rumpf, "code": totp(geheimnis, time.time())},
|
||||
headers=kopf,
|
||||
timeout=10,
|
||||
)
|
||||
pruefe("gesperrt gilt auch fuer die richtige Anmeldung (429)", richtig.status_code == 429, f"HTTP {richtig.status_code}")
|
||||
|
||||
print("== Anmeldeprotokoll")
|
||||
protokoll = ablage / "anmeldungen.jsonl"
|
||||
if not protokoll.is_file():
|
||||
pruefe("Anmeldeprotokoll vorhanden", False, str(protokoll))
|
||||
else:
|
||||
text = protokoll.read_text(encoding="utf-8")
|
||||
pruefe("Protokoll traegt das Passwort nicht", passwort not in text and "ganz-falsches-passwort" not in text, "Passwort im Protokoll")
|
||||
pruefe("Protokoll traegt den Code nicht", code not in text, "Code im Protokoll")
|
||||
ergebnisse = {json.loads(z).get("ergebnis") for z in text.splitlines() if z.strip()}
|
||||
pruefe("Protokoll kennt ok und falsch", {"ok", "falsch"} <= ergebnisse, str(ergebnisse))
|
||||
|
||||
print()
|
||||
print(f"{len(bestanden)} Faelle ok, {len(befunde)} Befund(e).")
|
||||
for befund in befunde:
|
||||
print(f"BEFUND: {befund}")
|
||||
return 2 if befunde else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
sys.exit(main())
|
||||
except Exception as ausnahme: # noqa: BLE001 - ein Absturz des Pruefskripts ist selbst ein Befund
|
||||
for befund in befunde:
|
||||
print(f"BEFUND: {befund}")
|
||||
print(f"BEFUND: das Pruefskript ist abgebrochen: {ausnahme!r}")
|
||||
sys.exit(2)
|
||||
Reference in new issue
Block a user