GuidoandClaude Opus 5
7a3bd200b1
Seit PSD2 muessen EU-Banken eine Schnittstelle anbieten, aber nur fuer lizenzierte Dienste mit eIDAS-Zertifikat. Enable Banking haelt die Lizenz und reicht sie weiter: eigene Anwendung registrieren, eigene Konten verknuepfen, danach genau diese lesen. Deckt N26 und Revolut ab; die Schweiz faellt nicht unter PSD2, Raiffeisen bleibt beim CSV-Weg. Es gibt keinen API-Schluessel. Jede Anfrage traegt ein selbst gebautes JWT, signiert mit dem privaten Schluessel der Anwendung (RS256). Signiert wird mit cryptography, das ohnehin schon fuer die Verschluesselung im Projekt ist – der Test prueft die Signatur echt gegen den oeffentlichen Schluessel, denn genau daran scheitert so eine Anbindung sonst als Erstes. Abgerufene Umsaetze werden NICHT gebucht. Sie laufen durch detect_rows und landen als Vorschlag im Eingang – dieselbe Strecke wie Mail und Kurzbefehl, also greifen Duplikatschutz und Schlagwortregeln ohne Sonderbehandlung. Eine dritte Quelle, kein dritter Sonderweg. Bewusst eng gehalten: nur Lesen; Zuordnung Bankkonto -> Konto trifft der Nutzer; die Zuordnung verlangt dieselbe Waehrung; die Rueckleitungs-Adresse wird eingetragen statt geraten (hinter einem Proxy sieht der Dienst "http", obwohl der Browser HTTPS spricht) und muss HTTPS sein. Der Ablauf der Zustimmung (bis 180 Tage) wird sichtbar gemeldet, damit der Abruf nicht eines Tages stumm aufhoert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
361 lines
14 KiB
Python
361 lines
14 KiB
Python
"""Bankanbindung über PSD2 gegen einen Doppelgänger der Schnittstelle.
|
||
|
||
Ein echter Zugang wäre im Test weder verfügbar noch wünschenswert: Er hinge an
|
||
einer registrierten Anwendung, einem privaten Schlüssel und echten Bankkonten.
|
||
Der Doppelgänger unten spricht nur die vier Endpunkte, die ``bank_api``
|
||
benutzt.
|
||
|
||
Was hier NICHT gefälscht wird, ist die Kryptografie: Das JWT wird echt
|
||
signiert und gegen den öffentlichen Schlüssel geprüft. Genau daran scheitert
|
||
die Anbindung sonst als Erstes, und der Fehler wäre von aussen nicht zu sehen.
|
||
"""
|
||
|
||
import base64
|
||
import json
|
||
import secrets
|
||
|
||
import pytest
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
||
|
||
from app.services import bank_api, bank_sync
|
||
|
||
SCHLUESSEL = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||
PEM = SCHLUESSEL.private_bytes(
|
||
encoding=serialization.Encoding.PEM,
|
||
format=serialization.PrivateFormat.PKCS8,
|
||
encryption_algorithm=serialization.NoEncryption(),
|
||
).decode()
|
||
|
||
|
||
class FakeBank:
|
||
"""Minimale Gegenstelle: nur was ``BankClient`` tatsächlich aufruft."""
|
||
|
||
def __init__(self, umsaetze=None, konten=None, seiten=1):
|
||
self.umsaetze = umsaetze or []
|
||
self.konten = konten if konten is not None else [{
|
||
"uid": "acc-1", "name": "Girokonto", "currency": "EUR",
|
||
"account_id": {"iban": "DE02120300000000202051"},
|
||
}]
|
||
self.seiten = seiten
|
||
self.aufrufe: list[str] = []
|
||
self.jwt: str | None = None
|
||
|
||
def __call__(self, methode, pfad, headers=None, json=None, **kw):
|
||
self.jwt = (headers or {}).get("Authorization", "")[7:]
|
||
self.aufrufe.append(f"{methode} {pfad.split('?')[0]}")
|
||
if pfad.startswith("/aspsps"):
|
||
return 200, {"aspsps": [{"name": "N26", "country": "DE"},
|
||
{"name": "Revolut", "country": "LT"}]}
|
||
if pfad == "/auth":
|
||
return 200, {"url": "https://bank.example/authorize?x=1",
|
||
"authorization_id": "auth-1"}
|
||
if pfad == "/sessions":
|
||
if json.get("code") != "guter-code":
|
||
return 400, {"message": "Code ungültig"}
|
||
return 200, {"session_id": "sess-1", "accounts": self.konten,
|
||
"access": {"valid_until": "2027-01-01T00:00:00+00:00"}}
|
||
if "/balances" in pfad:
|
||
return 200, {"balances": [{"balance_amount": {"amount": "100.00",
|
||
"currency": "EUR"}}]}
|
||
if "/transactions" in pfad:
|
||
# Blättern nachstellen: erst mit Fortsetzungsschlüssel, dann ohne.
|
||
hat_schluessel = "continuation_key=" in pfad
|
||
if self.seiten > 1 and not hat_schluessel:
|
||
return 200, {"transactions": self.umsaetze[:1], "continuation_key": "s2"}
|
||
rest = self.umsaetze[1:] if self.seiten > 1 else self.umsaetze
|
||
return 200, {"transactions": rest}
|
||
return 404, {"message": "unbekannt"}
|
||
|
||
|
||
@pytest.fixture()
|
||
def fake_bank(monkeypatch):
|
||
"""Ersetzt den HTTP-Aufruf; gibt eine Funktion zum Bestücken zurück."""
|
||
|
||
def bestuecken(**kw):
|
||
gegenstelle = FakeBank(**kw)
|
||
|
||
def _anfrage(self, methode, pfad, **rest):
|
||
code, daten = gegenstelle(methode, pfad, **rest)
|
||
if code >= 400:
|
||
raise bank_api.BankFehler(daten.get("message", "Fehler"))
|
||
return daten
|
||
|
||
monkeypatch.setattr(bank_api.BankClient, "_anfrage", _anfrage)
|
||
return gegenstelle
|
||
|
||
return bestuecken
|
||
|
||
|
||
def _app(admin, **kw):
|
||
body = {"application_id": "app-123", "private_key": PEM,
|
||
"base_url": "https://certain.example.com"}
|
||
body.update(kw)
|
||
r = admin.put("/api/bank/app", json=body)
|
||
assert r.status_code == 200, r.text
|
||
return r.json()
|
||
|
||
|
||
def _konto(admin, **kw):
|
||
body = {"name": "N26 Konto", "type": "girokonto", "currency": "EUR",
|
||
"opening_balance_cents": 0}
|
||
body.update(kw)
|
||
return admin.post("/api/accounts", json=body).json()
|
||
|
||
|
||
UMSATZ = {
|
||
"entry_reference": "tx-1",
|
||
"booking_date": "2026-08-20",
|
||
"transaction_amount": {"amount": "14.90", "currency": "EUR"},
|
||
"credit_debit_indicator": "DBIT",
|
||
"creditor": {"name": "SPEEDEE RESTAURATIONS"},
|
||
"remittance_information": ["Kartenzahlung"],
|
||
}
|
||
|
||
|
||
# --- Zugangsdaten ------------------------------------------------------------
|
||
|
||
|
||
def test_schluessel_wird_verschluesselt_abgelegt_und_nie_ausgeliefert(admin):
|
||
from app.database import SessionLocal
|
||
from app.models import BankApp
|
||
from app.services import geheimnis
|
||
|
||
ausgabe = _app(admin)
|
||
assert "private_key" not in ausgabe
|
||
assert ausgabe["key_set"] is True
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
zeile = db.query(BankApp).order_by(BankApp.id.desc()).first()
|
||
assert "PRIVATE KEY" not in zeile.private_key_enc, "Klartext in der Datenbank"
|
||
assert geheimnis.entschluesseln(zeile.private_key_enc) == PEM
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_rueckleitung_ohne_https_wird_abgelehnt(admin):
|
||
"""Banken leiten nicht auf eine unverschlüsselte Adresse zurück."""
|
||
r = admin.put("/api/bank/app", json={
|
||
"application_id": "app-123", "private_key": PEM,
|
||
"base_url": "http://192.168.1.50:8071"})
|
||
assert r.status_code == 422
|
||
assert "https" in r.text.lower()
|
||
|
||
|
||
def test_ohne_hinterlegte_anwendung_kommt_eine_klare_ansage(admin):
|
||
admin.delete("/api/bank/app")
|
||
r = admin.get("/api/bank/banks?country=DE")
|
||
assert r.status_code == 400
|
||
assert "Anwendung" in r.json()["detail"]
|
||
|
||
|
||
# --- JWT ---------------------------------------------------------------------
|
||
|
||
|
||
def test_jwt_ist_echt_signiert_und_nachpruefbar():
|
||
"""Die Gegenstelle prüft die Signatur – wenn sie hier stimmt, stimmt sie dort."""
|
||
jwt = bank_api.baue_jwt("app-123", PEM)
|
||
kopf_b64, rumpf_b64, sig_b64 = jwt.split(".")
|
||
pad = lambda s: s + "=" * (-len(s) % 4)
|
||
|
||
kopf = json.loads(base64.urlsafe_b64decode(pad(kopf_b64)))
|
||
assert kopf == {"typ": "JWT", "alg": "RS256", "kid": "app-123"}
|
||
rumpf = json.loads(base64.urlsafe_b64decode(pad(rumpf_b64)))
|
||
assert rumpf["iss"] == "enablebanking.com"
|
||
assert rumpf["aud"] == "api.enablebanking.com"
|
||
assert rumpf["exp"] - rumpf["iat"] == 3600
|
||
|
||
SCHLUESSEL.public_key().verify(
|
||
base64.urlsafe_b64decode(pad(sig_b64)),
|
||
f"{kopf_b64}.{rumpf_b64}".encode(),
|
||
padding.PKCS1v15(), hashes.SHA256())
|
||
|
||
|
||
def test_unbrauchbarer_schluessel_meldet_sich_verstaendlich():
|
||
with pytest.raises(bank_api.BankFehler) as fehler:
|
||
bank_api.baue_jwt("app-123", "das ist kein schluessel")
|
||
assert "pem" in str(fehler.value).lower()
|
||
|
||
|
||
# --- Verbinden ---------------------------------------------------------------
|
||
|
||
|
||
def test_banken_eines_landes_abrufen(admin, fake_bank):
|
||
_app(admin)
|
||
fake_bank()
|
||
r = admin.get("/api/bank/banks?country=DE")
|
||
assert r.status_code == 200, r.text
|
||
assert [b["name"] for b in r.json()] == ["N26", "Revolut"]
|
||
|
||
|
||
def test_verbinden_und_konten_finden(admin, fake_bank):
|
||
_app(admin)
|
||
gegenstelle = fake_bank()
|
||
|
||
start = admin.post("/api/bank/connect", json={"aspsp_name": "N26", "country": "DE"})
|
||
assert start.status_code == 200, start.text
|
||
assert start.json()["url"].startswith("https://")
|
||
state = start.json()["state"]
|
||
|
||
fertig = admin.post(f"/api/bank/callback?state={state}&code=guter-code")
|
||
assert fertig.status_code == 200, fertig.text
|
||
assert fertig.json()["konten"] == 1
|
||
|
||
verknuepfungen = [v for v in admin.get("/api/bank/links").json()
|
||
if v["remote_uid"] == "acc-1"]
|
||
assert len(verknuepfungen) == 1
|
||
assert verknuepfungen[0]["iban"] == "DE02120300000000202051"
|
||
# Noch keinem Konto zugeordnet – das entscheidet der Nutzer.
|
||
assert verknuepfungen[0]["account_id"] is None
|
||
assert "POST /auth" in gegenstelle.aufrufe and "POST /sessions" in gegenstelle.aufrufe
|
||
|
||
|
||
def test_falscher_code_wird_verstaendlich_abgewiesen(admin, fake_bank):
|
||
_app(admin)
|
||
fake_bank()
|
||
state = admin.post("/api/bank/connect",
|
||
json={"aspsp_name": "N26", "country": "DE"}).json()["state"]
|
||
r = admin.post(f"/api/bank/callback?state={state}&code=falsch")
|
||
assert r.status_code == 400
|
||
assert "ungültig" in r.json()["detail"]
|
||
|
||
|
||
def test_fremder_state_findet_keine_verbindung(admin, fake_bank):
|
||
_app(admin)
|
||
fake_bank()
|
||
r = admin.post("/api/bank/callback?state=ausgedacht&code=guter-code")
|
||
assert r.status_code == 400
|
||
assert "keine offene Verbindung" in r.json()["detail"]
|
||
|
||
|
||
# --- Kontozuordnung ----------------------------------------------------------
|
||
|
||
|
||
def _verbunden(admin, fake_bank_fn, **kw):
|
||
"""Verbindung mit EIGENER Konto-Kennung je Aufruf.
|
||
|
||
Die Zuordnung eines Bankkontos ist absichtlich dauerhaft: Verbindet man
|
||
dieselbe Bank neu, bleibt sie erhalten. In der gemeinsamen Testdatenbank
|
||
würde derselbe uid deshalb die Zuordnung eines früheren Tests erben.
|
||
"""
|
||
_app(admin)
|
||
kw.setdefault("konten", [{
|
||
"uid": f"acc-{secrets.token_hex(4)}", "name": "Girokonto",
|
||
"currency": "EUR", "account_id": {"iban": "DE02120300000000202051"},
|
||
}])
|
||
gegenstelle = fake_bank_fn(**kw)
|
||
state = admin.post("/api/bank/connect",
|
||
json={"aspsp_name": "N26", "country": "DE"}).json()["state"]
|
||
antwort = admin.post(f"/api/bank/callback?state={state}&code=guter-code").json()
|
||
link = [v for v in admin.get("/api/bank/links").json()
|
||
if v["connection_id"] == antwort["connection_id"]][0]
|
||
return antwort["connection_id"], link, gegenstelle
|
||
|
||
|
||
def test_zuordnung_verlangt_dieselbe_waehrung(admin, fake_bank):
|
||
"""Ein EUR-Konto in ein CHF-Konto zu schreiben verfälscht jeden Saldo."""
|
||
_, link, _ = _verbunden(admin, fake_bank)
|
||
chf = _konto(admin, name="Franken-Konto", currency="CHF")
|
||
r = admin.post(f"/api/bank/links/{link['id']}/account/{chf['id']}")
|
||
assert r.status_code == 400
|
||
assert "Währungen passen nicht" in r.json()["detail"]
|
||
|
||
eur = _konto(admin, name="Euro-Konto", currency="EUR")
|
||
ok = admin.post(f"/api/bank/links/{link['id']}/account/{eur['id']}")
|
||
assert ok.status_code == 200, ok.text
|
||
assert ok.json()["account_id"] == eur["id"]
|
||
|
||
|
||
# --- Umsätze holen -----------------------------------------------------------
|
||
|
||
|
||
def test_ohne_zuordnung_wird_nichts_geholt(admin, fake_bank):
|
||
verbindung, _, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ])
|
||
r = admin.post(f"/api/bank/connections/{verbindung}/sync")
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["ohne_zuordnung"] is True
|
||
assert r.json()["neu"] == 0
|
||
|
||
|
||
def test_umsatz_landet_als_vorschlag_im_eingang(admin, fake_bank):
|
||
"""Nichts wird gebucht – die Bank liefert Vorschläge wie Mail und Kurzbefehl."""
|
||
verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ])
|
||
konto = _konto(admin, name="N26 fuer Abruf", currency="EUR")
|
||
admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}")
|
||
|
||
r = admin.post(f"/api/bank/connections/{verbindung}/sync")
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["neu"] == 1
|
||
|
||
treffer = [e for e in admin.get("/api/inbox").json()
|
||
if e["payee"] == "SPEEDEE RESTAURATIONS"]
|
||
assert len(treffer) == 1
|
||
# DBIT ist eine Ausgabe – das Vorzeichen steckt in der Richtung, nicht im Betrag.
|
||
assert treffer[0]["amount_cents"] == -1490
|
||
assert treffer[0]["date"] == "2026-08-20"
|
||
assert treffer[0]["source"] == "bank"
|
||
assert treffer[0]["status"] == "pending"
|
||
assert treffer[0]["transaction_id"] is None
|
||
# Und der Saldo hat sich nicht bewegt.
|
||
assert admin.get(f"/api/accounts/{konto['id']}").json()["balance_cents"] == 0
|
||
|
||
|
||
def test_gutschrift_kommt_als_einnahme_an(admin, fake_bank):
|
||
gutschrift = dict(UMSATZ, entry_reference="tx-crdt",
|
||
credit_debit_indicator="CRDT",
|
||
creditor=None, debtor={"name": "Arbeitgeber AG"})
|
||
verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[gutschrift])
|
||
konto = _konto(admin, name="N26 Einnahme", currency="EUR")
|
||
admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}")
|
||
admin.post(f"/api/bank/connections/{verbindung}/sync")
|
||
|
||
treffer = [e for e in admin.get("/api/inbox").json()
|
||
if e["payee"] == "Arbeitgeber AG"]
|
||
assert treffer and treffer[0]["amount_cents"] == 1490
|
||
|
||
|
||
def test_zweiter_abruf_holt_nichts_doppelt(admin, fake_bank):
|
||
"""Die Kennung des Umsatzes zählt – sonst wüchse der Eingang bei jedem Abruf."""
|
||
verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=[UMSATZ])
|
||
konto = _konto(admin, name="N26 zweimal", currency="EUR")
|
||
admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}")
|
||
|
||
erst = admin.post(f"/api/bank/connections/{verbindung}/sync").json()
|
||
zweit = admin.post(f"/api/bank/connections/{verbindung}/sync").json()
|
||
assert erst["neu"] == 1
|
||
assert zweit["neu"] == 0
|
||
|
||
|
||
def test_blaettern_holt_alle_seiten(admin, fake_bank):
|
||
"""Ohne die Fortsetzungs-Schleife käme still nur die erste Seite an."""
|
||
umsaetze = [dict(UMSATZ, entry_reference=f"tx-{i}") for i in range(2)]
|
||
verbindung, link, _ = _verbunden(admin, fake_bank, umsaetze=umsaetze, seiten=2)
|
||
konto = _konto(admin, name="N26 blaettern", currency="EUR")
|
||
admin.post(f"/api/bank/links/{link['id']}/account/{konto['id']}")
|
||
|
||
assert admin.post(f"/api/bank/connections/{verbindung}/sync").json()["neu"] == 2
|
||
|
||
|
||
def test_ablauf_der_zustimmung_wird_gemeldet(admin, fake_bank):
|
||
"""Läuft die Zustimmung ab, muss das sichtbar sein statt still zu enden."""
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
from app.database import SessionLocal
|
||
from app.models import BankConnection
|
||
|
||
verbindung, _, _ = _verbunden(admin, fake_bank)
|
||
liste = [c for c in admin.get("/api/bank/connections").json() if c["id"] == verbindung]
|
||
assert liste and liste[0]["expires_soon"] is False
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
zeile = db.get(BankConnection, verbindung)
|
||
zeile.valid_until = datetime.now(timezone.utc) + timedelta(days=3)
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
|
||
liste = [c for c in admin.get("/api/bank/connections").json() if c["id"] == verbindung]
|
||
assert liste[0]["expires_soon"] is True
|