certain/backend/app/services/bank_api.py

"""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}."