diff --git a/backend/app/main.py b/backend/app/main.py index 887587e..95db894 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -28,6 +28,7 @@ from .migrate import upgrade_to_head log = logging.getLogger(__name__) from .routers import ( accounts, + bank, auth, backup, budgets, @@ -159,6 +160,7 @@ api.include_router(rates.router) api.include_router(quick.router) api.include_router(inbox.router) api.include_router(mail.router) +api.include_router(bank.router) api.include_router(uebertraege.router) # Fehler an den Kurzbefehl-Endpunkten zusätzlich als „message" ausliefern. diff --git a/backend/app/models.py b/backend/app/models.py index 9976279..a82f716 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -629,6 +629,7 @@ class ApiToken(Base): class InboxSource(str, enum.Enum): shortcut = "shortcut" # per Kurzbefehl vom Telefon erfasst mail = "mail" # aus einer Benachrichtigungs-Mail gelesen + bank = "bank" # direkt von der Bank geholt (PSD2) class InboxStatus(str, enum.Enum): @@ -831,3 +832,84 @@ class LinkIgnored(Base): __table_args__ = ( UniqueConstraint("a_txn_id", "b_txn_id", name="uq_ignored_pair"), ) + + +class BankApp(Base): + """Die bei Enable Banking registrierte Anwendung dieser Installation. + + Es gibt keinen API-Schlüssel zum Eintippen: Enable Banking gibt eine + Application ID und eine private Schlüsseldatei aus, mit der jede Anfrage + einzeln signiert wird (JWT, RS256). Der Schlüssel liegt deshalb hier – + verschlüsselt wie die IMAP-Passwörter, denn er ist genauso schützenswert. + + Wer Certain selbst hostet, registriert eine eigene Anwendung; mitliefern + lässt sich das nicht, weil der Schlüssel an die Registrierung gebunden ist. + """ + + __tablename__ = "bank_apps" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + owner_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + application_id: Mapped[str] = mapped_column(String(64), nullable=False) + private_key_enc: Mapped[str] = mapped_column(Text, nullable=False) + # Öffentliche Adresse dieser Installation, an die die Bank zurückleitet. + # Muss HTTPS sein und im Control Panel hinterlegt werden. Bewusst + # eingetragen statt geraten: Hinter einem Proxy sieht der Dienst „http", + # obwohl der Browser HTTPS spricht. + base_url: Mapped[str] = mapped_column(String(255), nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) + + +class BankConnection(Base): + """Eine autorisierte Verbindung zu einer Bank – das Ergebnis von /sessions.""" + + __tablename__ = "bank_connections" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + owner_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + aspsp_name: Mapped[str] = mapped_column(String(120), nullable=False) + aspsp_country: Mapped[str] = mapped_column(String(2), nullable=False) + session_id: Mapped[str | None] = mapped_column(String(64), nullable=True) + # Der Zeitpunkt, ab dem neu autorisiert werden muss (bis zu 180 Tage). + # Sichtbar anzeigen: Läuft er ab, hört der Abruf sonst stumm auf. + valid_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + # Solange die Autorisierung läuft, steht hier der state-Wert aus /auth. + pending_state: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + last_sync_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) + + +class BankAccountLink(Base): + """Ein Konto bei der Bank, zugeordnet zu einem Konto in Certain. + + Ohne Zuordnung wird nichts abgerufen – dieselbe Regel wie bei den + Mail-Absendern: Certain rät nicht, wohin eine Bewegung gehört. + """ + + __tablename__ = "bank_account_links" + + id: Mapped[int] = mapped_column(Integer, primary_key=True) + owner_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True + ) + connection_id: Mapped[int] = mapped_column( + ForeignKey("bank_connections.id", ondelete="CASCADE"), nullable=False, index=True + ) + # Kennung des Kontos bei Enable Banking; damit werden Salden und Umsätze geholt. + remote_uid: Mapped[str] = mapped_column(String(64), nullable=False) + iban: Mapped[str | None] = mapped_column(String(64), nullable=True) + remote_name: Mapped[str | None] = mapped_column(String(255), nullable=True) + currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR") + account_id: Mapped[int | None] = mapped_column( + ForeignKey("accounts.id", ondelete="SET NULL"), nullable=True + ) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now) + + __table_args__ = ( + UniqueConstraint("owner_id", "remote_uid", name="uq_bank_link_remote"), + ) diff --git a/backend/app/routers/bank.py b/backend/app/routers/bank.py new file mode 100644 index 0000000..151a074 --- /dev/null +++ b/backend/app/routers/bank.py @@ -0,0 +1,170 @@ +"""Bankkonten über PSD2 verbinden und abrufen.""" + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from ..database import get_db +from ..deps import get_current_user, get_owned +from ..models import Account, BankAccountLink, BankApp, BankConnection, User +from ..schemas import ( + BankAccountLinkOut, + BankAppIn, + BankAppOut, + BankConnectionOut, + BankConnectStart, +) +from ..services import bank_sync, geheimnis +from ..services.bank_api import BankFehler + +router = APIRouter(prefix="/bank", tags=["bank"]) + + +def _fehler(e: BankFehler) -> HTTPException: + return HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +# --- Die Anwendung dieser Installation --------------------------------------- + + +@router.get("/app", response_model=BankAppOut | None) +def get_app(db: Session = Depends(get_db), user: User = Depends(get_current_user)): + return db.query(BankApp).filter(BankApp.owner_id == user.id).first() + + +@router.put("/app", response_model=BankAppOut) +def set_app(payload: BankAppIn, db: Session = Depends(get_db), + user: User = Depends(get_current_user)): + """Application ID, Schlüsseldatei und Rückleitungs-Adresse hinterlegen.""" + app = db.query(BankApp).filter(BankApp.owner_id == user.id).first() + if app is None: + app = BankApp(owner_id=user.id) + db.add(app) + app.application_id = payload.application_id.strip() + app.private_key_enc = geheimnis.verschluesseln(payload.private_key) + app.base_url = payload.base_url + db.commit() + db.refresh(app) + return app + + +@router.delete("/app", status_code=status.HTTP_204_NO_CONTENT) +def delete_app(db: Session = Depends(get_db), user: User = Depends(get_current_user)): + app = db.query(BankApp).filter(BankApp.owner_id == user.id).first() + if app is not None: + db.delete(app) + db.commit() + + +@router.get("/banks") +def list_banks(country: str, db: Session = Depends(get_db), + user: User = Depends(get_current_user)) -> list[dict]: + """Verfügbare Banken eines Landes – für die Auswahl beim Verbinden.""" + try: + return bank_sync.banken(db, user, country) + except BankFehler as e: + raise _fehler(e) from e + + +# --- Verbindungen ------------------------------------------------------------ + + +@router.get("/connections", response_model=list[BankConnectionOut]) +def list_connections(db: Session = Depends(get_db), user: User = Depends(get_current_user)): + verbindungen = ( + db.query(BankConnection) + .filter(BankConnection.owner_id == user.id, BankConnection.session_id.isnot(None)) + .order_by(BankConnection.id) + .all() + ) + out = [] + for v in verbindungen: + eintrag = BankConnectionOut.model_validate(v) + eintrag.expires_soon = bank_sync.laeuft_bald_ab(v) + out.append(eintrag) + return out + + +@router.post("/connect") +def connect(payload: BankConnectStart, db: Session = Depends(get_db), + user: User = Depends(get_current_user)) -> dict: + """Autorisierung starten. Die Antwort enthält die Adresse der Bank.""" + try: + return bank_sync.starte_verbindung( + db, user, payload.aspsp_name, payload.country, payload.psu_type) + except BankFehler as e: + raise _fehler(e) from e + + +@router.post("/callback") +def callback(state: str, code: str, db: Session = Depends(get_db), + user: User = Depends(get_current_user)) -> dict: + """Der Code von der Rückleitung wird zur Sitzung samt Kontoliste.""" + try: + return bank_sync.schliesse_verbindung(db, user, state, code) + except BankFehler as e: + raise _fehler(e) from e + + +@router.post("/connections/{connection_id}/sync") +def sync(connection_id: int, db: Session = Depends(get_db), + user: User = Depends(get_current_user)) -> dict: + verbindung = get_owned(db, BankConnection, connection_id, user) + try: + return bank_sync.hole_umsaetze(db, user, verbindung) + except BankFehler as e: + verbindung.last_error = str(e)[:500] + db.commit() + raise _fehler(e) from e + + +@router.delete("/connections/{connection_id}", status_code=status.HTTP_204_NO_CONTENT) +def delete_connection(connection_id: int, db: Session = Depends(get_db), + user: User = Depends(get_current_user)): + db.delete(get_owned(db, BankConnection, connection_id, user)) + db.commit() + + +# --- Kontozuordnung ---------------------------------------------------------- + + +@router.get("/links", response_model=list[BankAccountLinkOut]) +def list_links(db: Session = Depends(get_db), user: User = Depends(get_current_user)): + return ( + db.query(BankAccountLink) + .filter(BankAccountLink.owner_id == user.id) + .order_by(BankAccountLink.id) + .all() + ) + + +@router.post("/links/{link_id}/account/{account_id}", response_model=BankAccountLinkOut) +def set_link_account(link_id: int, account_id: int, db: Session = Depends(get_db), + user: User = Depends(get_current_user)): + """Ein Bankkonto einem Konto in Certain zuordnen. + + Ohne Zuordnung wird nichts abgerufen – Certain rät nicht, wohin fremde + Umsätze gehören. + """ + verknuepfung = get_owned(db, BankAccountLink, link_id, user) + konto = get_owned(db, Account, account_id, user) + if konto.currency.upper() != verknuepfung.currency.upper(): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=(f"Währungen passen nicht: Das Bankkonto führt " + f"{verknuepfung.currency}, das Konto '{konto.name}' führt " + f"{konto.currency}."), + ) + verknuepfung.account_id = konto.id + db.commit() + db.refresh(verknuepfung) + return verknuepfung + + +@router.delete("/links/{link_id}/account", response_model=BankAccountLinkOut) +def clear_link_account(link_id: int, db: Session = Depends(get_db), + user: User = Depends(get_current_user)): + verknuepfung = get_owned(db, BankAccountLink, link_id, user) + verknuepfung.account_id = None + db.commit() + db.refresh(verknuepfung) + return verknuepfung diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 1f2e47b..ef784d8 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -593,3 +593,72 @@ class MailSenderOut(MailSenderIn): source_id: int model_config = ConfigDict(from_attributes=True) + + +# --- Bankanbindung (PSD2) ---------------------------------------------------- + + +class BankAppIn(BaseModel): + application_id: str = Field(min_length=1, max_length=64) + # Inhalt der .pem-Datei aus dem Control Panel. + private_key: str = Field(min_length=1) + # Öffentliche Adresse dieser Installation; die Bank leitet dorthin zurück. + base_url: str = Field(min_length=1, max_length=255) + + @field_validator("base_url") + @classmethod + def nur_https(cls, v: str) -> str: + """Ohne HTTPS lehnt die Bank die Rückleitung ab. + + Bewusst eingetragen statt erkannt: Hinter einem Proxy sieht der Dienst + selbst „http", obwohl der Browser HTTPS spricht – automatisches Raten + würde hier zuverlässig das Falsche liefern. + """ + v = v.strip().rstrip("/") + if not v.lower().startswith("https://"): + raise ValueError( + "Die Adresse muss mit https:// beginnen. Banken leiten nicht " + "auf eine unverschlüsselte Adresse zurück." + ) + return v + + +class BankAppOut(BaseModel): + id: int + application_id: str + base_url: str + created_at: datetime | None + # Der Schlüssel verlässt den Server nie – nur die Tatsache, dass einer da ist. + key_set: bool = True + + model_config = ConfigDict(from_attributes=True) + + +class BankConnectionOut(BaseModel): + id: int + aspsp_name: str + aspsp_country: str + valid_until: datetime | None + last_sync_at: datetime | None + last_error: str | None + expires_soon: bool = False + + model_config = ConfigDict(from_attributes=True) + + +class BankAccountLinkOut(BaseModel): + id: int + connection_id: int + remote_uid: str + iban: str | None + remote_name: str | None + currency: str + account_id: int | None + + model_config = ConfigDict(from_attributes=True) + + +class BankConnectStart(BaseModel): + aspsp_name: str = Field(min_length=1, max_length=120) + country: str = Field(min_length=2, max_length=2) + psu_type: Literal["personal", "business"] = "personal" diff --git a/backend/app/services/backup_data.py b/backend/app/services/backup_data.py index 725695c..95ad1c6 100644 --- a/backend/app/services/backup_data.py +++ b/backend/app/services/backup_data.py @@ -149,6 +149,15 @@ USER_TABLES: list[TableSpec] = [ # ``pending_uploads`` – dort liegen rohe Dateien, die noch niemand geprüft hat. # Sie sind flüchtig, blähen ein JSON-Backup um Megabytes auf und lassen sich # jederzeit erneut vom Telefon schicken. +# +# ``bank_apps``/``bank_connections``/``bank_account_links`` – dieselbe +# Überlegung wie beim Postfach, nur schärfer: ``bank_apps`` trägt den privaten +# Schlüssel, mit dem jede Anfrage an die Bank signiert wird, und +# ``bank_connections`` eine laufende Sitzung. Beides gehört in keine Datei, die +# man weitergibt. Die Sitzung liesse sich ohnehin nicht wiederbeleben – die +# Zustimmung hängt an der Bank, nicht an unserer Datenbank. Nach einem +# Wiedereinspielen verbindet man die Bank neu; die abgerufenen Buchungen selbst +# sind längst gesichert. # Nur im systemweiten Backup, je Benutzer. SYSTEM_USER_TABLES: list[TableSpec] = [ diff --git a/backend/app/services/bank_api.py b/backend/app/services/bank_api.py new file mode 100644 index 0000000..0bd7aeb --- /dev/null +++ b/backend/app/services/bank_api.py @@ -0,0 +1,173 @@ +"""Zugriff auf Bankkonten über PSD2 (Enable Banking). + +Seit PSD2 müssen EU-Banken eine Schnittstelle anbieten, aber nur für +lizenzierte Dienste mit eIDAS-Zertifikat. Enable Banking hält diese Lizenz und +reicht sie weiter: Man registriert eine eigene Anwendung, verknüpft die eigenen +Konten und darf danach genau diese lesen („Restricted Mode"). + +Es gibt keinen API-Schlüssel. Jede Anfrage trägt ein selbst gebautes JWT, +signiert mit dem privaten Schlüssel der Anwendung (RS256, eine Stunde gültig). +Signiert wird mit ``cryptography``, das ohnehin schon für die Verschlüsselung +der Zugangsdaten im Projekt ist – keine zusätzliche Abhängigkeit. + +Nur Lesen: Salden und Umsätze. Zahlungen auslösen kann diese Anbindung nicht, +und das soll auch so bleiben. +""" + +from __future__ import annotations + +import base64 +import json +import logging +import time +from datetime import date, datetime, timedelta, timezone + +import httpx +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import padding, rsa + +log = logging.getLogger(__name__) + +BASIS = "https://api.enablebanking.com" +JWT_GUELTIG_SEKUNDEN = 3600 +ZEITLIMIT = 30.0 +# Wie weit ein erster Abruf zurückgeht. Mehr geben die meisten Banken über +# PSD2 ohnehin nicht heraus. +ERSTER_ABRUF_TAGE = 90 + + +class BankFehler(RuntimeError): + """Fehler beim Zugriff – mit einer Ursache, die man dem Nutzer zeigen kann.""" + + +def _b64(roh: bytes) -> str: + """URL-sicheres Base64 ohne Füllzeichen, wie es JWT verlangt.""" + return base64.urlsafe_b64encode(roh).rstrip(b"=").decode("ascii") + + +def baue_jwt(application_id: str, private_key_pem: str) -> str: + """Signiertes JWT für eine Stunde. + + Kopf und Rumpf sind von Enable Banking vorgegeben; ``kid`` ist die + Application ID, an der die Gegenseite den öffentlichen Schlüssel findet. + """ + try: + schluessel = serialization.load_pem_private_key( + private_key_pem.encode("utf-8"), password=None + ) + except (ValueError, TypeError) as e: + raise BankFehler( + "Der private Schlüssel lässt sich nicht lesen. Ist es die " + "unveränderte .pem-Datei aus dem Control Panel?" + ) from e + if not isinstance(schluessel, rsa.RSAPrivateKey): + raise BankFehler("Der Schlüssel ist kein RSA-Schlüssel.") + + jetzt = int(time.time()) + kopf = {"typ": "JWT", "alg": "RS256", "kid": application_id} + rumpf = { + "iss": "enablebanking.com", + "aud": "api.enablebanking.com", + "iat": jetzt, + "exp": jetzt + JWT_GUELTIG_SEKUNDEN, + } + zu_signieren = f"{_b64(json.dumps(kopf).encode())}.{_b64(json.dumps(rumpf).encode())}" + signatur = schluessel.sign( + zu_signieren.encode("ascii"), padding.PKCS1v15(), hashes.SHA256() + ) + return f"{zu_signieren}.{_b64(signatur)}" + + +class BankClient: + """Dünne Hülle um die vier Endpunkte, die wir brauchen. + + Als Klasse und nicht als Funktionssammlung, damit das JWT je Abruf-Runde + einmal gebaut wird statt je Anfrage – und damit die Tests einen + Doppelgänger einsetzen können, ohne dass ein Netzzugang nötig ist. + """ + + def __init__(self, application_id: str, private_key_pem: str, basis: str = BASIS): + self._jwt = baue_jwt(application_id, private_key_pem) + self._basis = basis.rstrip("/") + + def _anfrage(self, methode: str, pfad: str, **kw) -> dict: + kopfzeilen = {"Authorization": f"Bearer {self._jwt}", + "Content-Type": "application/json"} + try: + with httpx.Client(timeout=ZEITLIMIT) as c: + antwort = c.request(methode, f"{self._basis}{pfad}", + headers=kopfzeilen, **kw) + except httpx.HTTPError as e: + raise BankFehler(f"Die Bank-Schnittstelle ist nicht erreichbar: {e}") from e + if antwort.status_code >= 400: + raise BankFehler(_fehlertext(antwort)) + try: + return antwort.json() + except ValueError as e: + raise BankFehler("Unerwartete Antwort der Schnittstelle.") from e + + # --- Die vier Endpunkte ------------------------------------------------- + + def banken(self, land: str) -> list[dict]: + """Verfügbare Banken eines Landes – für die Auswahl beim Verbinden.""" + daten = self._anfrage("GET", f"/aspsps?country={land.upper()}") + return daten.get("aspsps", []) + + def starte_autorisierung(self, aspsp_name: str, land: str, redirect_url: str, + state: str, tage: int, psu_type: str = "personal") -> dict: + """Weiterleitungs-Adresse für die Zustimmung bei der Bank.""" + gueltig_bis = datetime.now(timezone.utc) + timedelta(days=tage) + return self._anfrage("POST", "/auth", json={ + "access": {"valid_until": gueltig_bis.isoformat()}, + "aspsp": {"name": aspsp_name, "country": land.upper()}, + "redirect_url": redirect_url, + "state": state, + "psu_type": psu_type, + }) + + def sitzung(self, code: str) -> dict: + """Den Rückleitungs-Code gegen eine Sitzung samt Kontoliste tauschen.""" + return self._anfrage("POST", "/sessions", json={"code": code}) + + def salden(self, konto_uid: str) -> list[dict]: + return self._anfrage("GET", f"/accounts/{konto_uid}/balances").get("balances", []) + + def umsaetze(self, konto_uid: str, seit: date) -> list[dict]: + """Alle Umsätze ab einem Datum – über alle Seiten hinweg. + + Die Schnittstelle blättert über einen ``continuation_key``. Ohne diese + Schleife bekäme man stillschweigend nur die erste Seite; bei einem + Jahresauszug wäre der Rest einfach weg. + """ + gesammelt: list[dict] = [] + weiter: str | None = None + for _ in range(50): # Bremse gegen eine endlose Antwortkette + pfad = f"/accounts/{konto_uid}/transactions?date_from={seit.isoformat()}" + if weiter: + pfad += f"&continuation_key={weiter}" + daten = self._anfrage("GET", pfad) + gesammelt.extend(daten.get("transactions", [])) + weiter = daten.get("continuation_key") + if not weiter: + break + else: + log.warning("Umsatzabruf nach 50 Seiten abgebrochen (Konto %s)", konto_uid) + return gesammelt + + +def _fehlertext(antwort: httpx.Response) -> str: + """Aus der Fehlerantwort etwas machen, das im Formular hilft.""" + if antwort.status_code == 401: + return ("Die Anmeldung wurde abgelehnt. Passen Application ID und " + "Schlüsseldatei zusammen, und ist die Anwendung aktiv?") + if antwort.status_code == 403: + return ("Zugriff verweigert. Im eingeschränkten Betrieb sind nur die " + "Konten lesbar, die du selbst verknüpft hast.") + try: + inhalt = antwort.json() + grund = inhalt.get("message") or inhalt.get("error") or inhalt.get("detail") + if grund: + return f"Die Bank-Schnittstelle meldet: {grund}" + except ValueError: + pass + return f"Die Bank-Schnittstelle antwortete mit Fehler {antwort.status_code}." diff --git a/backend/app/services/bank_sync.py b/backend/app/services/bank_sync.py new file mode 100644 index 0000000..c190271 --- /dev/null +++ b/backend/app/services/bank_sync.py @@ -0,0 +1,289 @@ +"""Bankkonten verbinden und Umsätze abholen. + +Der Weg in drei Schritten, genau wie ihn die Schnittstelle vorgibt: + + 1. ``starte_verbindung`` → Adresse, zu der der Nutzer geschickt wird + 2. ``schliesse_verbindung`` → der Rückleitungs-Code wird zur Sitzung, + und wir erfahren, welche Konten es dort gibt + 3. ``hole_umsaetze`` → Umsätze je zugeordnetem Konto + +Abgeholte Umsätze werden NICHT gebucht. Sie landen als Vorschlag im Eingang, +dieselbe Strecke wie Mails und Kurzbefehl-Erfassungen – und laufen dabei durch +``detect_rows``, sodass Duplikatschutz und Schlagwortregeln ohne +Sonderbehandlung greifen. Eine dritte Quelle, kein dritter Sonderweg. +""" + +from __future__ import annotations + +import logging +import secrets +from datetime import date, datetime, timedelta, timezone + +from sqlalchemy.orm import Session + +from ..models import ( + Account, + BankAccountLink, + BankApp, + BankConnection, + InboxItem, + InboxSource, + InboxStatus, + User, +) +from . import csv_import, geheimnis +from .bank_api import ERSTER_ABRUF_TAGE, BankClient, BankFehler + +log = logging.getLogger(__name__) + +# Wie lange die Zustimmung gelten soll. Die Bank kürzt selbst, wenn sie weniger +# erlaubt; mehr als 180 Tage lässt die Regelung nicht zu. +ZUSTIMMUNG_TAGE = 180 +# Ab wann gewarnt wird, dass neu autorisiert werden muss. Läuft die Zustimmung +# ab, hört der Abruf sonst stumm auf – dieselbe Falle wie „0 neu" ohne Grund. +WARNUNG_TAGE = 14 + + +def _client(db: Session, user: User) -> BankClient: + app = db.query(BankApp).filter(BankApp.owner_id == user.id).first() + if app is None: + raise BankFehler( + "Es ist noch keine Bank-Anwendung hinterlegt. Trag Application ID " + "und Schlüsseldatei in den Einstellungen ein." + ) + schluessel = geheimnis.entschluesseln(app.private_key_enc) + if schluessel is None: + raise BankFehler( + "Der gespeicherte Schlüssel lässt sich nicht mehr entschlüsseln. " + "Wurde das Server-Geheimnis getauscht? Dann bitte neu hinterlegen." + ) + return BankClient(app.application_id, schluessel) + + +def banken(db: Session, user: User, land: str) -> list[dict]: + """Banken eines Landes, auf das Nötige eingedampft.""" + roh = _client(db, user).banken(land) + return [ + {"name": b.get("name"), "country": b.get("country"), + "logo": b.get("logo"), "psu_types": b.get("psu_types", [])} + for b in roh if b.get("name") + ] + + +def starte_verbindung(db: Session, user: User, aspsp_name: str, land: str, + psu_type: str = "personal") -> dict: + """Autorisierung anstossen und die Weiterleitungs-Adresse liefern.""" + app = db.query(BankApp).filter(BankApp.owner_id == user.id).first() + if app is None: + raise BankFehler("Es ist noch keine Bank-Anwendung hinterlegt.") + + state = secrets.token_urlsafe(24) + verbindung = BankConnection( + owner_id=user.id, aspsp_name=aspsp_name, aspsp_country=land.upper(), + pending_state=state, + ) + db.add(verbindung) + db.flush() + + antwort = _client(db, user).starte_autorisierung( + aspsp_name, land, f"{app.base_url.rstrip('/')}/bank/rueckkehr", + state, ZUSTIMMUNG_TAGE, psu_type, + ) + db.commit() + return {"url": antwort.get("url"), "connection_id": verbindung.id, "state": state} + + +def schliesse_verbindung(db: Session, user: User, state: str, code: str) -> dict: + """Code gegen die Sitzung tauschen und die gefundenen Konten ablegen. + + Die Konten werden nur VERZEICHNET, nicht zugeordnet: Welches Bankkonto zu + welchem Konto in Certain gehört, entscheidet der Nutzer. Raten wäre hier + besonders teuer, weil eine falsche Zuordnung fremde Umsätze in ein Konto + schreiben würde. + """ + verbindung = ( + db.query(BankConnection) + .filter(BankConnection.owner_id == user.id, BankConnection.pending_state == state) + .first() + ) + if verbindung is None: + raise BankFehler("Zu dieser Rückmeldung gibt es keine offene Verbindung.") + + daten = _client(db, user).sitzung(code) + verbindung.session_id = daten.get("session_id") + verbindung.pending_state = None + verbindung.last_error = None + gueltig = (daten.get("access") or {}).get("valid_until") + if gueltig: + try: + verbindung.valid_until = datetime.fromisoformat(gueltig.replace("Z", "+00:00")) + except ValueError: + verbindung.valid_until = None + + neu = 0 + for konto in daten.get("accounts", []): + uid = konto.get("uid") + if not uid: + continue + vorhanden = ( + db.query(BankAccountLink) + .filter(BankAccountLink.owner_id == user.id, BankAccountLink.remote_uid == uid) + .first() + ) + if vorhanden is not None: + vorhanden.connection_id = verbindung.id + continue + db.add(BankAccountLink( + owner_id=user.id, + connection_id=verbindung.id, + remote_uid=uid, + iban=(konto.get("account_id") or {}).get("iban"), + remote_name=konto.get("name"), + currency=(konto.get("currency") or "EUR").upper(), + )) + neu += 1 + + db.commit() + return {"connection_id": verbindung.id, "konten": neu, + "valid_until": verbindung.valid_until.isoformat() if verbindung.valid_until else None} + + +def _betrag_cents(umsatz: dict) -> int | None: + """Vorzeichenbehafteter Betrag in Cent. + + PSD2 führt den Betrag ohne Vorzeichen und die Richtung getrennt in + ``credit_debit_indicator`` (CRDT/DBIT). Manche Banken liefern trotzdem ein + Vorzeichen mit – beides wird berücksichtigt, damit eine Ausgabe nicht als + Einnahme ankommt. + """ + betrag = (umsatz.get("transaction_amount") or {}).get("amount") + if betrag is None: + return None + cents = csv_import.parse_amount(str(betrag), ".") + if cents is None: + return None + richtung = (umsatz.get("credit_debit_indicator") or "").upper() + if richtung == "DBIT": + return -abs(cents) + if richtung == "CRDT": + return abs(cents) + return cents + + +def _text(umsatz: dict, ausgabe: bool) -> tuple[str | None, str | None]: + """Empfänger und Verwendungszweck. Bei einer Ausgabe zählt der Empfänger, + bei einer Einnahme der Absender.""" + gegen = umsatz.get("creditor") if ausgabe else umsatz.get("debtor") + name = (gegen or {}).get("name") + zweck = umsatz.get("remittance_information") + if isinstance(zweck, list): + zweck = " ".join(str(z) for z in zweck if z) + return (str(name)[:255] if name else None, + str(zweck)[:255] if zweck else None) + + +def hole_umsaetze(db: Session, user: User, verbindung: BankConnection) -> dict: + """Umsätze aller zugeordneten Konten dieser Verbindung in den Eingang holen.""" + verknuepfungen = [ + v for v in db.query(BankAccountLink).filter( + BankAccountLink.connection_id == verbindung.id, + BankAccountLink.owner_id == user.id, + ) + if v.account_id is not None + ] + if not verknuepfungen: + return {"geholt": 0, "neu": 0, "ohne_zuordnung": True} + + client = _client(db, user) + konten = {a.id: a for a in db.query(Account).filter(Account.owner_id == user.id)} + seit = (verbindung.last_sync_at.date() - timedelta(days=3) + if verbindung.last_sync_at + else date.today() - timedelta(days=ERSTER_ABRUF_TAGE)) + + geholt = neu = 0 + for verknuepfung in verknuepfungen: + konto = konten.get(verknuepfung.account_id) + if konto is None: + continue + for umsatz in client.umsaetze(verknuepfung.remote_uid, seit): + geholt += 1 + kennung = (umsatz.get("entry_reference") or umsatz.get("transaction_id") + or umsatz.get("uid")) + if not kennung: + continue # ohne Kennung kein Schutz vor Doppelabruf + kennung = f"{verknuepfung.remote_uid}:{kennung}"[:255] + schon = ( + db.query(InboxItem.id) + .filter(InboxItem.owner_id == user.id, + InboxItem.source == InboxSource.bank, + InboxItem.external_id == kennung) + .first() + ) + if schon is not None: + continue + + cents = _betrag_cents(umsatz) + tag = _datum(umsatz) + if cents is None or cents == 0 or tag is None: + continue + payee, zweck = _text(umsatz, cents < 0) + + # Durch dieselbe Erkennung wie CSV und Kurzbefehl: Schlagwortregeln + # schlagen die Kategorie vor, der Eingangs-Abgleich verhindert, dass + # eine bereits erfasste Zahlung ein zweites Mal auftaucht. + erkannt = csv_import.detect_rows(db, user, konto, [{ + "date": tag.isoformat(), "amount_cents": cents, + "payee": payee, "purpose": zweck, + }]) + zeile = erkannt[0] if erkannt else {} + if zeile.get("is_duplicate"): + continue + + db.add(InboxItem( + owner_id=user.id, + source=InboxSource.bank, + status=InboxStatus.pending, + account_id=konto.id, + date=tag, + amount_cents=cents, + currency=(umsatz.get("transaction_amount") or {}).get("currency") + or konto.currency, + payee=payee, + note=zweck, + category_id=zeile.get("category_id"), + external_id=kennung, + )) + neu += 1 + + verbindung.last_sync_at = datetime.now(timezone.utc) + verbindung.last_error = None + db.commit() + return {"geholt": geholt, "neu": neu, "ohne_zuordnung": False} + + +def _datum(umsatz: dict) -> date | None: + for feld in ("booking_date", "value_date", "transaction_date"): + roh = umsatz.get(feld) + if not roh: + continue + try: + return date.fromisoformat(str(roh)[:10]) + except ValueError: + continue + return None + + +def laeuft_bald_ab(verbindung: BankConnection) -> bool: + """Läuft die Zustimmung demnächst ab? + + SQLite gibt Zeitstempel ohne Zeitzone zurück, auch wenn die Spalte als + ``timezone=True`` deklariert ist. Ohne diese Ergänzung scheitert der + Vergleich mit einem TypeError – und zwar erst beim Anzeigen der + Verbindungsliste, also weit weg von der Ursache. + """ + if verbindung.valid_until is None: + return False + bis = verbindung.valid_until + if bis.tzinfo is None: + bis = bis.replace(tzinfo=timezone.utc) + return bis - datetime.now(timezone.utc) < timedelta(days=WARNUNG_TAGE) diff --git a/backend/migrations/versions/0021_bank_anbindung.py b/backend/migrations/versions/0021_bank_anbindung.py new file mode 100644 index 0000000..da8bbdb --- /dev/null +++ b/backend/migrations/versions/0021_bank_anbindung.py @@ -0,0 +1,76 @@ +"""Anbindung an Banken ueber PSD2 (Enable Banking). + +Drei Ebenen: die bei Enable Banking registrierte Anwendung dieser Installation +(Application ID + privater Schluessel), die autorisierte Verbindung zu einer +Bank, und die Zuordnung eines dortigen Kontos zu einem Konto in Certain. + +Der private Schluessel liegt verschluesselt – er ist so schuetzenswert wie ein +Passwort, denn mit ihm wird jede Anfrage signiert. + +Revision ID: 0021 +Revises: 0020 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "0021" +down_revision = "0020" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "bank_apps", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("owner_id", sa.Integer(), nullable=False), + sa.Column("application_id", sa.String(length=64), nullable=False), + sa.Column("private_key_enc", sa.Text(), nullable=False), + sa.Column("base_url", sa.String(length=255), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + ) + op.create_index("ix_bank_apps_owner_id", "bank_apps", ["owner_id"]) + + op.create_table( + "bank_connections", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("owner_id", sa.Integer(), nullable=False), + sa.Column("aspsp_name", sa.String(length=120), nullable=False), + sa.Column("aspsp_country", sa.String(length=2), nullable=False), + sa.Column("session_id", sa.String(length=64), nullable=True), + sa.Column("valid_until", sa.DateTime(timezone=True), nullable=True), + sa.Column("pending_state", sa.String(length=64), nullable=True), + sa.Column("last_sync_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + ) + op.create_index("ix_bank_connections_owner_id", "bank_connections", ["owner_id"]) + op.create_index("ix_bank_connections_state", "bank_connections", ["pending_state"]) + + op.create_table( + "bank_account_links", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("owner_id", sa.Integer(), nullable=False), + sa.Column("connection_id", sa.Integer(), nullable=False), + sa.Column("remote_uid", sa.String(length=64), nullable=False), + sa.Column("iban", sa.String(length=64), nullable=True), + sa.Column("remote_name", sa.String(length=255), nullable=True), + sa.Column("currency", sa.String(length=3), nullable=False, server_default="EUR"), + sa.Column("account_id", sa.Integer(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["owner_id"], ["users.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["connection_id"], ["bank_connections.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"], ondelete="SET NULL"), + sa.UniqueConstraint("owner_id", "remote_uid", name="uq_bank_link_remote"), + ) + op.create_index("ix_bank_links_owner_id", "bank_account_links", ["owner_id"]) + op.create_index("ix_bank_links_connection", "bank_account_links", ["connection_id"]) + + +def downgrade() -> None: + op.drop_table("bank_account_links") + op.drop_table("bank_connections") + op.drop_table("bank_apps") diff --git a/backend/tests/test_backup.py b/backend/tests/test_backup.py index 94fc916..3b530f2 100644 --- a/backend/tests/test_backup.py +++ b/backend/tests/test_backup.py @@ -73,7 +73,8 @@ def test_jede_nutzertabelle_ist_im_backup_beruecksichtigt(): for spec in liste } # Bewusst ausgenommen – die Begründung steht in backup_data.py. - ausgenommen = {"mail_sources", "mail_senders", "pending_uploads"} + ausgenommen = {"mail_sources", "mail_senders", "pending_uploads", + "bank_apps", "bank_connections", "bank_account_links"} fehlend = sorted( m.class_.__tablename__ diff --git a/backend/tests/test_bank.py b/backend/tests/test_bank.py new file mode 100644 index 0000000..94f5b60 --- /dev/null +++ b/backend/tests/test_bank.py @@ -0,0 +1,360 @@ +"""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