GuidoandClaude Opus 5
f3a08f2937
Wer Bank-Mails in ein Sammelpostfach weiterleitet, hat im From je nach Methode die eigene Adresse stehen. Geprueft werden deshalb auch Reply-To, Return-Path, Sender und X-Forwarded-For – bei automatischer Weiterleitung bleibt der Originalabsender dort erhalten. Ausserdem meldet der Abruf jetzt, welche Mails er uebergangen hat, mit Absender und Betreff. Beim Einrichten war das sonst Raten: Man sah nur "3 uebersprungen" und wusste nicht, was in den Filter gehoert. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
252 lines
9.2 KiB
Python
252 lines
9.2 KiB
Python
"""Postfächer, Verschlüsselung und der Abruf gegen einen IMAP-Doppelgänger.
|
||
|
||
Ein echter Mailserver wäre im Test weder verfügbar noch wünschenswert. Der
|
||
Doppelgänger unten spricht nur die Handvoll IMAP-Befehle, die ``mail_import``
|
||
benutzt, und liefert die Muster-Mails aus ``fixtures/mail`` aus.
|
||
"""
|
||
|
||
import pathlib
|
||
|
||
import pytest
|
||
|
||
from app.services import geheimnis, mail_import
|
||
|
||
MUSTER = pathlib.Path(__file__).parent / "fixtures" / "mail"
|
||
|
||
|
||
class FakeImap:
|
||
"""Minimaler IMAP-Server: nur was ``mail_import`` tatsächlich aufruft."""
|
||
|
||
def __init__(self, mails: list[bytes]):
|
||
self.mails = mails
|
||
self.gesehen: list[bytes] = []
|
||
self.kopiert: list[tuple[bytes, str]] = []
|
||
self.abgemeldet = False
|
||
|
||
def login(self, user, passwort):
|
||
return "OK", [b""]
|
||
|
||
def list(self):
|
||
return "OK", [b'(\\HasNoChildren) "/" "INBOX"', b'(\\HasNoChildren) "/" "Belege"']
|
||
|
||
def select(self, ordner):
|
||
return ("OK", [b"1"]) if "INBOX" in ordner or "Belege" in ordner else ("NO", [b""])
|
||
|
||
def search(self, charset, kriterium):
|
||
return "OK", [b" ".join(str(i + 1).encode() for i in range(len(self.mails)))]
|
||
|
||
def fetch(self, uid, teile):
|
||
return "OK", [(b"1 (RFC822 {1})", self.mails[int(uid) - 1])]
|
||
|
||
def store(self, uid, befehl, flags):
|
||
if "Seen" in flags:
|
||
self.gesehen.append(uid)
|
||
return "OK", [b""]
|
||
|
||
def copy(self, uid, ordner):
|
||
self.kopiert.append((uid, ordner))
|
||
return "OK", [b""]
|
||
|
||
def logout(self):
|
||
self.abgemeldet = True
|
||
return "BYE", [b""]
|
||
|
||
|
||
@pytest.fixture()
|
||
def fake_imap(monkeypatch):
|
||
"""Ersetzt die Verbindung; gibt eine Funktion zum Bestücken zurück."""
|
||
gehalten = {}
|
||
|
||
def bestuecken(*dateinamen):
|
||
mails = [(MUSTER / n).read_bytes() for n in dateinamen]
|
||
server = FakeImap(mails)
|
||
gehalten["server"] = server
|
||
monkeypatch.setattr(mail_import, "_verbinden", lambda quelle: server)
|
||
return server
|
||
|
||
return bestuecken
|
||
|
||
|
||
def _konto(admin, **kw):
|
||
body = {"name": "Raiffeisen", "type": "girokonto", "currency": "CHF",
|
||
"opening_balance_cents": 0}
|
||
body.update(kw)
|
||
return admin.post("/api/accounts", json=body).json()
|
||
|
||
|
||
def _postfach(admin, **kw):
|
||
body = {"name": "Belege", "host": "imap.example.com", "port": 993,
|
||
"username": "kundin@example.com", "password": "geheim123", "folder": "INBOX"}
|
||
body.update(kw)
|
||
r = admin.post("/api/mail/sources", json=body)
|
||
assert r.status_code == 201, r.text
|
||
return r.json()
|
||
|
||
|
||
def test_passwort_wird_verschluesselt_abgelegt_und_nie_ausgeliefert(admin):
|
||
from app.database import SessionLocal
|
||
from app.models import MailSource
|
||
|
||
quelle = _postfach(admin)
|
||
assert "password" not in quelle
|
||
assert quelle["password_set"] is True
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
zeile = db.get(MailSource, quelle["id"])
|
||
assert zeile.password_enc != "geheim123", "Klartext in der Datenbank"
|
||
assert geheimnis.entschluesseln(zeile.password_enc) == "geheim123"
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_unlesbares_passwort_meldet_sich_verstaendlich(admin, monkeypatch):
|
||
"""Getauschtes Server-Geheimnis darf nicht in einem Stacktrace enden."""
|
||
from app.database import SessionLocal
|
||
from app.models import MailSource
|
||
|
||
quelle = _postfach(admin)
|
||
db = SessionLocal()
|
||
try:
|
||
zeile = db.get(MailSource, quelle["id"])
|
||
zeile.password_enc = "kaputt"
|
||
db.commit()
|
||
with pytest.raises(mail_import.MailFehler) as fehler:
|
||
mail_import._verbinden(zeile)
|
||
assert "entschlüsseln" in str(fehler.value)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_passwort_bleibt_wenn_es_beim_aendern_weggelassen_wird(admin):
|
||
from app.database import SessionLocal
|
||
from app.models import MailSource
|
||
|
||
quelle = _postfach(admin)
|
||
r = admin.put(f"/api/mail/sources/{quelle['id']}", json={
|
||
"host": "imap.example.com", "username": "kundin@example.com", "folder": "Belege"})
|
||
assert r.status_code == 200, r.text
|
||
assert r.json()["folder"] == "Belege"
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
assert geheimnis.entschluesseln(db.get(MailSource, quelle["id"]).password_enc) == "geheim123"
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def test_abruf_legt_erkannte_zahlung_in_den_eingang(admin, fake_imap):
|
||
konto = _konto(admin)
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"name": "Raiffeisen", "from_contains": "raiffeisen.ch", "account_id": konto["id"]})
|
||
server = fake_imap("raiffeisen_zahlungsstatus.eml")
|
||
|
||
r = admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
|
||
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"] == "Sunrise GmbH / Yallo"]
|
||
assert len(treffer) == 1
|
||
assert treffer[0]["amount_cents"] == -1490
|
||
assert treffer[0]["currency"] == "CHF"
|
||
assert treffer[0]["date"] == "2026-07-31"
|
||
assert treffer[0]["account_id"] == konto["id"]
|
||
# Nichts wird stillschweigend gebucht.
|
||
assert treffer[0]["status"] == "pending"
|
||
assert treffer[0]["transaction_id"] is None
|
||
assert server.gesehen, "Mail muss als gelesen markiert werden"
|
||
|
||
|
||
def test_dieselbe_mail_zweimal_ergibt_einen_eintrag(admin, fake_imap):
|
||
"""Die Message-ID zählt, nicht die UID – die vergibt der Server neu."""
|
||
konto = _konto(admin)
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
|
||
fake_imap("raiffeisen_zahlungsstatus.eml")
|
||
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
|
||
vorher = len(admin.get("/api/inbox").json())
|
||
zweiter = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
|
||
assert zweiter["neu"] == 0
|
||
assert zweiter["uebersprungen"] == 1
|
||
assert len(admin.get("/api/inbox").json()) == vorher
|
||
|
||
|
||
def test_mail_ohne_passende_zuordnung_wird_uebergangen(admin, fake_imap):
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "sparkasse.de"})
|
||
fake_imap("raiffeisen_zahlungsstatus.eml")
|
||
|
||
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
|
||
assert ergebnis["neu"] == 0
|
||
assert ergebnis["uebersprungen"] == 1
|
||
|
||
|
||
def test_zuordnung_ohne_konto_landet_als_unklar(admin, fake_imap):
|
||
"""Ohne Zielkonto lässt sich nicht buchen – das muss sichtbar sein."""
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "revolut.com"})
|
||
fake_imap("revolut_ueberweisung.eml")
|
||
|
||
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["unklar"] == 1
|
||
|
||
|
||
def test_verbindungstest_liefert_die_ordner(admin, fake_imap):
|
||
quelle = _postfach(admin)
|
||
fake_imap()
|
||
antwort = admin.post(f"/api/mail/sources/{quelle['id']}/test")
|
||
assert antwort.status_code == 200, antwort.text
|
||
assert "Belege" in antwort.json()["folders"]
|
||
|
||
|
||
def test_verschieben_statt_nur_markieren(admin, fake_imap):
|
||
konto = _konto(admin)
|
||
quelle = _postfach(admin, done_folder="Belege")
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
|
||
server = fake_imap("raiffeisen_zahlungsstatus.eml")
|
||
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/fetch")
|
||
assert server.kopiert and "Belege" in server.kopiert[0][1]
|
||
|
||
|
||
def test_weitergeleitete_mail_wird_am_originalabsender_erkannt(admin, fake_imap, monkeypatch):
|
||
"""Leitet man Bank-Mails in ein Sammelpostfach weiter, steht die eigene
|
||
Adresse im ``From``. Der Originalabsender bleibt aber in den
|
||
Weiterleitungs-Koepfen erhalten – daran muss die Zuordnung greifen."""
|
||
konto = _konto(admin)
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={
|
||
"from_contains": "raiffeisen.ch", "account_id": konto["id"]})
|
||
|
||
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes()
|
||
weitergeleitet = roh.replace(
|
||
b"From: noreply@ebanking.raiffeisen.ch",
|
||
b"From: ich@gmail.com\nX-Forwarded-For: noreply@ebanking.raiffeisen.ch ich@gmail.com",
|
||
).replace(b"<raiffeisen-muster-1@", b"<weitergeleitet-1@")
|
||
|
||
server = FakeImap([weitergeleitet])
|
||
monkeypatch.setattr(mail_import, "_verbinden", lambda q: server)
|
||
assert admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()["neu"] == 1
|
||
|
||
|
||
def test_uebergangene_mails_nennen_absender_und_betreff(admin, monkeypatch):
|
||
"""Beim Einrichten muss man sehen, was in den Filter gehoert."""
|
||
quelle = _postfach(admin)
|
||
admin.post(f"/api/mail/sources/{quelle['id']}/senders", json={"from_contains": "passt.nie"})
|
||
# Eigene Message-ID: die Testdatenbank lebt ueber den ganzen Lauf, sonst
|
||
# gilt die Mail als bereits erfasst und die Zuordnung wird nie geprueft.
|
||
roh = (MUSTER / "raiffeisen_zahlungsstatus.eml").read_bytes().replace(
|
||
b"<raiffeisen-muster-1@", b"<ohne-zuordnung-1@")
|
||
monkeypatch.setattr(mail_import, "_verbinden", lambda q: FakeImap([roh]))
|
||
|
||
ergebnis = admin.post(f"/api/mail/sources/{quelle['id']}/fetch").json()
|
||
assert ergebnis["uebersprungen"] == 1
|
||
assert "raiffeisen.ch" in ergebnis["ohne_zuordnung"][0]["from"]
|
||
assert "Zahlungsstatus" in ergebnis["ohne_zuordnung"][0]["subject"]
|