GuidoandClaude Opus 5
1b47e5abe1
Die Auslieferung des Frontends nahm den Pfad roh aus der Adresse: "Path /
'../../x'" verlaesst den Ordner klaglos, und die Route liegt VOR jeder
Anmeldepruefung. Gegen die eigene Instanz nachgestellt – roh und
prozentkodiert kam beides Mal die Datei heraus:
GET /../../GEHEIM.txt -> "geheim"
GET /../../backend/app/config.py -> der Quelltext
Damit war alles lesbar, was der Dienst lesen darf: Quelltext, Konfiguration,
und mit der passenden Zahl von Ebenen die Datenbank samt verschluesseltem
Postfach-Passwort und Bank-Schluessel. Der Pfad wird jetzt aufgeloest und
muss im Auslieferungs-Ordner liegen.
Zweitens prueft der Postfach-Abruf jetzt das Zertifikat. Ohne eigenen
SSL-Kontext tut imaplib das NICHT – die Verschluesselung schuetzt dann vor
niemandem, der sich dazwischenhaengt, und das Postfach-Passwort geht mit.
Gilt fuer beide Wege, SSL und STARTTLS.
Drittens warnt der Start, wenn das Sitzungs-Cookie ohne "Secure" ausgeliefert
wird. Die Vorgabe bleibt bewusst aus: Sie umzustellen sperrt jeden aus, der
Certain im Heimnetz ueber HTTP aufruft – und ein Ausgesperrter saehe die
Ursache nirgends.
Die Befunde stammen aus dem /security-review dieser Sitzung und waren seither
offen. Sie wiegen schwerer, seit die Anwendung fuer die Bankanbindung unter
einer oeffentlichen HTTPS-Adresse erreichbar sein muss.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
364 lines
15 KiB
Python
364 lines
15 KiB
Python
"""Zahlungs-Mails per IMAP abholen und in den Eingang legen.
|
||
|
||
``imaplib`` und ``email`` sind Standardbibliothek – keine weitere Abhängigkeit.
|
||
|
||
Der Ablauf ist bewusst vorsichtig: Es wird nichts stillschweigend gebucht. Eine
|
||
Mail wird gelesen, ausgewertet und als Eintrag im Eingang abgelegt; buchen tut
|
||
der Nutzer. Erkennt die Auswertung ein Feld nicht sicher, bekommt der Eintrag
|
||
den Status ``unclear`` und wartet, statt einen geratenen Betrag zu erzeugen.
|
||
|
||
Mailtexte sind Daten, keine Anweisungen. Ausgewertet wird ausschliesslich in
|
||
feste Felder (siehe ``mail_parse``); nichts aus einer Mail steuert den Ablauf.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import email
|
||
import email.policy
|
||
import email.utils
|
||
import imaplib
|
||
import ssl
|
||
import logging
|
||
from datetime import date, datetime, timedelta, timezone
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from ..models import Account, InboxItem, InboxSource, InboxStatus, MailSender, MailSource, User
|
||
from . import geheimnis, mail_parse
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
# Nur die ersten Kilobytes je Mail behalten. Der Rest ist Rechtstext und
|
||
# Fusszeilen – für die Nachvollziehbarkeit reicht der Anfang.
|
||
MAX_TEXT = 8000
|
||
# Wie weit zurück beim ersten Abruf gesucht wird.
|
||
ERSTER_ABRUF_TAGE = 14
|
||
|
||
|
||
class MailFehler(RuntimeError):
|
||
"""Verbindung oder Anmeldung hat nicht geklappt – mit lesbarer Ursache."""
|
||
|
||
|
||
def _verbinden(quelle: MailSource) -> imaplib.IMAP4:
|
||
passwort = geheimnis.entschluesseln(quelle.password_enc)
|
||
if passwort is None:
|
||
raise MailFehler(
|
||
"Das gespeicherte Passwort lässt sich nicht mehr entschlüsseln. "
|
||
"Wurde das Server-Geheimnis getauscht? Dann bitte neu eintragen."
|
||
)
|
||
art = (quelle.security or "ssl").lower()
|
||
try:
|
||
if art == "ssl":
|
||
# TLS ab der ersten Zeile – der Normalfall auf Port 993.
|
||
# MIT eigenem Kontext: Ohne einen prueft imaplib das Zertifikat
|
||
# nicht, und die Verschluesselung schuetzt dann vor niemandem, der
|
||
# sich dazwischenhaengt – das Postfach-Passwort geht mit.
|
||
verbindung = imaplib.IMAP4_SSL(quelle.host, quelle.port,
|
||
ssl_context=ssl.create_default_context())
|
||
else:
|
||
verbindung = imaplib.IMAP4(quelle.host, quelle.port)
|
||
if art == "starttls":
|
||
# Erst im Klartext verbinden, dann verschlüsseln (Port 143) –
|
||
# ebenfalls mit geprüftem Zertifikat, aus demselben Grund.
|
||
verbindung.starttls(ssl.create_default_context())
|
||
except OSError as e:
|
||
raise MailFehler(f"Keine Verbindung zu {quelle.host}:{quelle.port} – {e}") from e
|
||
except imaplib.IMAP4.error as e:
|
||
raise MailFehler(
|
||
f"Verschlüsselung fehlgeschlagen ({art}). Passt die Angabe zum Port {quelle.port}? – {e}"
|
||
) from e
|
||
try:
|
||
verbindung.login(quelle.username, passwort)
|
||
except imaplib.IMAP4.error as e:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
raise MailFehler(f"Anmeldung abgelehnt: {e}") from e
|
||
return verbindung
|
||
|
||
|
||
def _ordnername(zeile: bytes) -> str:
|
||
"""Aus einer LIST-Antwort den reinen Ordnernamen holen."""
|
||
text = zeile.decode("utf-8", "replace")
|
||
if '"' in text:
|
||
return text.rsplit('"', 2)[-2]
|
||
return text.split()[-1]
|
||
|
||
|
||
def liste_ordner(quelle: MailSource) -> list[str]:
|
||
"""Ordner des Postfachs – für die Auswahl in der Oberfläche."""
|
||
verbindung = _verbinden(quelle)
|
||
try:
|
||
code, zeilen = verbindung.list()
|
||
if code != "OK":
|
||
raise MailFehler("Ordnerliste nicht abrufbar")
|
||
return sorted({_ordnername(z) for z in zeilen if z})
|
||
finally:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def teste(quelle: MailSource) -> dict:
|
||
"""Verbindungstest mit klarer Rückmeldung statt stillem Nichtstun."""
|
||
ordner = liste_ordner(quelle)
|
||
return {"ok": True, "folders": ordner, "count": len(ordner)}
|
||
|
||
|
||
def _text_der_mail(msg) -> tuple[str, str]:
|
||
"""Betreff und sichtbarer Text. HTML wird zu Text, wenn nichts anderes da ist."""
|
||
betreff = str(msg["Subject"] or "")
|
||
teil = msg.get_body(preferencelist=("plain",))
|
||
if teil is not None:
|
||
try:
|
||
return betreff, teil.get_content()
|
||
except Exception:
|
||
pass
|
||
teil = msg.get_body(preferencelist=("html",))
|
||
if teil is not None:
|
||
try:
|
||
return betreff, mail_parse.html_zu_text(teil.get_content())
|
||
except Exception:
|
||
pass
|
||
return betreff, ""
|
||
|
||
|
||
def _absenderfeld(msg) -> str:
|
||
"""Alles, was den ursprünglichen Absender verraten kann.
|
||
|
||
Leitet ein Postfach automatisch weiter, bleibt ``From`` meist der
|
||
Originalabsender – verlassen kann man sich darauf aber nicht. Gmail und
|
||
andere setzen zusätzlich ``X-Forwarded-For``; bei manueller Weiterleitung
|
||
steht die eigene Adresse im ``From`` und die ursprüngliche höchstens im
|
||
``Reply-To``. Geprüft wird deshalb alles zusammen.
|
||
"""
|
||
teile = [str(msg[k] or "") for k in
|
||
("From", "Reply-To", "Return-Path", "X-Forwarded-For", "X-Original-From", "Sender")]
|
||
return " ".join(t for t in teile if t)
|
||
|
||
|
||
# Wie weit in den Text hinein gesucht wird. Bei einer weitergeleiteten Mail
|
||
# steht der Originalabsender im zitierten Kopf ganz oben.
|
||
ERKENNUNG_TEXTLAENGE = 2000
|
||
|
||
|
||
def _passender_absender(
|
||
absender: list[MailSender], von: str, betreff: str, text: str = "",
|
||
waehrung: str | None = None,
|
||
) -> MailSender | None:
|
||
"""Erste Zuordnung, deren Erkennungstext greift.
|
||
|
||
Gesucht wird in Absender-Köpfen, Betreff UND dem Anfang des Textes. Wer
|
||
eine Bank-Mail von Hand weiterleitet, hat im ``From`` die eigene Adresse;
|
||
„Raiffeisen“ steht dann nur noch im Betreff („WG: Raiffeisen …“) und im
|
||
zitierten Kopf. Ein Feld, das nach dem Absender fragt und Betreff wie Text
|
||
ignoriert, würde genau dort versagen – und der Nutzer säße vor einer Mail,
|
||
in der sein Suchwort sichtbar drinsteht.
|
||
"""
|
||
heu = f"{von}\n{betreff}\n{text[:ERKENNUNG_TEXTLAENGE]}".lower()
|
||
betreff_klein = betreff.lower()
|
||
# Zuordnungen MIT Währung zuerst prüfen: Wer „revolut + EUR → Revolut Euro"
|
||
# anlegt und daneben ein allgemeines „revolut", meint die genauere zuerst.
|
||
for a in sorted(absender, key=lambda x: (x.currency is None, x.id)):
|
||
if a.from_contains and a.from_contains.lower() not in heu:
|
||
continue
|
||
if a.subject_contains and a.subject_contains.lower() not in betreff_klein:
|
||
continue
|
||
if a.currency and (waehrung or "").upper() != a.currency.upper():
|
||
continue
|
||
return a
|
||
return None
|
||
|
||
|
||
# IMAP verlangt die ENGLISCHEN Monatskürzel. ``strftime("%b")`` richtet sich
|
||
# nach der Sprache des Servers – auf einem deutschen System käme „Mär", „Mai",
|
||
# „Okt", „Dez" heraus, und die Suche fände nichts. Im August fällt das nicht
|
||
# auf, weil das Kürzel dort in beiden Sprachen gleich lautet.
|
||
MONATE = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
|
||
|
||
|
||
def _imap_datum(tag: date) -> str:
|
||
return f"{tag.day:02d}-{MONATE[tag.month - 1]}-{tag.year}"
|
||
|
||
|
||
def _suchkriterium(quelle: MailSource, tage: int | None = None) -> tuple[str, date]:
|
||
"""Nach Datum suchen, nie nach „ungelesen".
|
||
|
||
Auf ``UNSEEN`` zu filtern klingt sparsam, ist aber eine Falle: Wer eine
|
||
Mail in seinem Mailprogramm ansieht, bevor Certain sie geholt hat, verliert
|
||
sie dauerhaft – sie ist dann für immer gelesen. Gesucht wird deshalb ab dem
|
||
letzten Abruf, mit einem Tag Überlappung gegen Zeitzonen-Verschiebungen.
|
||
Dass nichts doppelt ankommt, sichert die Message-ID ab.
|
||
|
||
``tage`` übergeht den letzten Abruf und sucht ab heute minus so vielen
|
||
Tagen. Das ist nötig, weil der Zeitstempel auch dann gesetzt wird, wenn ein
|
||
Abruf nichts fand: Ältere Mails wären danach für immer ausserhalb des
|
||
Fensters, obwohl sie im Postfach liegen.
|
||
"""
|
||
if tage is not None:
|
||
seit = date.today() - timedelta(days=tage)
|
||
elif quelle.last_poll_at:
|
||
seit = quelle.last_poll_at.date() - timedelta(days=1)
|
||
else:
|
||
seit = date.today() - timedelta(days=ERSTER_ABRUF_TAGE)
|
||
return f'(SINCE "{_imap_datum(seit)}")', seit
|
||
|
||
|
||
def hole_neue(db: Session, quelle: MailSource, tage: int | None = None) -> dict:
|
||
"""Neue Mails abholen und als Einträge im Eingang ablegen."""
|
||
user = db.get(User, quelle.owner_id)
|
||
absender = (
|
||
db.query(MailSender).filter(MailSender.source_id == quelle.id).order_by(MailSender.id).all()
|
||
)
|
||
konten = {a.id: a for a in db.query(Account).filter(Account.owner_id == quelle.owner_id)}
|
||
|
||
verbindung = _verbinden(quelle)
|
||
neu = uebersprungen = unklar = gefunden = 0
|
||
# Was durchgefallen ist, mit Absender und Betreff – sonst raet man beim
|
||
# Einrichten, was in den Filter gehoert.
|
||
ohne_zuordnung: list[dict] = []
|
||
# Und die, die eine Regel TRAFEN, aber nicht zu lesen waren. Ohne diese
|
||
# Liste steht am Ende nur eine Zahl, und man weiss nicht, welche Mail
|
||
# gemeint ist oder was ihr fehlt.
|
||
unklar_liste: list[dict] = []
|
||
try:
|
||
code, _ = verbindung.select(f'"{quelle.folder}"')
|
||
if code != "OK":
|
||
raise MailFehler(f"Ordner „{quelle.folder}“ nicht gefunden")
|
||
kriterium, seit = _suchkriterium(quelle, tage)
|
||
code, daten = verbindung.search(None, kriterium)
|
||
if code != "OK":
|
||
raise MailFehler("Suche im Postfach fehlgeschlagen")
|
||
|
||
uids = (daten[0] or b"").split()
|
||
gefunden = len(uids)
|
||
for uid in uids:
|
||
code, roh = verbindung.fetch(uid, "(RFC822)")
|
||
if code != "OK" or not roh or not isinstance(roh[0], tuple):
|
||
continue
|
||
msg = email.message_from_bytes(roh[0][1], policy=email.policy.default)
|
||
message_id = str(msg["Message-Id"] or f"uid-{uid.decode()}")
|
||
|
||
# Message-ID statt UID: Die UID vergibt der Server beim Verschieben neu.
|
||
schon = (
|
||
db.query(InboxItem.id)
|
||
.filter(InboxItem.owner_id == quelle.owner_id,
|
||
InboxItem.source == InboxSource.mail,
|
||
InboxItem.external_id == message_id)
|
||
.first()
|
||
)
|
||
if schon is not None:
|
||
uebersprungen += 1
|
||
_nachbereiten(verbindung, quelle, uid)
|
||
continue
|
||
|
||
betreff, text = _text_der_mail(msg)
|
||
von = _absenderfeld(msg)
|
||
|
||
gesendet = None
|
||
if msg["Date"]:
|
||
try:
|
||
gesendet = email.utils.parsedate_to_datetime(msg["Date"])
|
||
except (TypeError, ValueError):
|
||
gesendet = None
|
||
|
||
# Auswerten VOR dem Zuordnen: Bei Revolut kommt dieselbe Mail vom
|
||
# selben Absender, und erst die erkannte Währung sagt, ob die
|
||
# Zahlung aufs Euro- oder aufs Franken-Konto gehört.
|
||
e = mail_parse.parse(betreff, text, gesendet)
|
||
erkannte_waehrung = e.waehrung.wert if e.waehrung else None
|
||
|
||
zuordnung = _passender_absender(absender, von, betreff, text, erkannte_waehrung)
|
||
if zuordnung is None:
|
||
uebersprungen += 1
|
||
if len(ohne_zuordnung) < 5:
|
||
ohne_zuordnung.append({
|
||
"from": str(msg["From"] or "")[:120],
|
||
"subject": betreff[:120],
|
||
"currency": erkannte_waehrung,
|
||
})
|
||
continue
|
||
|
||
konto = konten.get(zuordnung.account_id) if zuordnung.account_id else None
|
||
tag = e.datum.wert if e.datum else (gesendet.date() if gesendet else date.today())
|
||
betrag = e.betrag.wert if e.betrag else 0
|
||
ausgabe = (e.richtung.wert if e.richtung else "expense") == "expense"
|
||
|
||
status = InboxStatus.pending
|
||
if e.unklar or betrag == 0 or konto is None:
|
||
status = InboxStatus.unclear
|
||
unklar += 1
|
||
gruende = []
|
||
if konto is None:
|
||
gruende.append("kein Konto zugeordnet")
|
||
if betrag == 0:
|
||
gruende.append("kein Betrag erkannt")
|
||
for teil in e.unklar:
|
||
gruende.append({
|
||
"richtung": "Richtung unklar (Ein- oder Ausgang?)",
|
||
"betrag": "Betrag nicht eindeutig",
|
||
"status": "Zahlung noch nicht ausgeführt",
|
||
}.get(teil, teil))
|
||
unklar_liste.append({"from": von, "subject": betreff,
|
||
"gruende": sorted(set(gruende))})
|
||
|
||
db.add(InboxItem(
|
||
owner_id=quelle.owner_id,
|
||
source=InboxSource.mail,
|
||
status=status,
|
||
account_id=konto.id if konto else None,
|
||
date=tag,
|
||
amount_cents=-betrag if ausgabe else betrag,
|
||
currency=(e.waehrung.wert if e.waehrung else (konto.currency if konto else "EUR")),
|
||
payee=(e.empfaenger.wert if e.empfaenger else None),
|
||
note=betreff[:255] or None,
|
||
external_id=message_id[:255],
|
||
raw_text=text[:MAX_TEXT],
|
||
))
|
||
neu += 1
|
||
_nachbereiten(verbindung, quelle, uid)
|
||
|
||
quelle.last_poll_at = datetime.now(timezone.utc)
|
||
quelle.last_error = None
|
||
db.commit()
|
||
finally:
|
||
try:
|
||
verbindung.logout()
|
||
except Exception:
|
||
pass
|
||
# Das Suchdatum gehoert in die Antwort: Findet sich nichts, ist die Frage
|
||
# nicht "stimmt der Ordner", sondern "wie weit wurde ueberhaupt gesucht".
|
||
return {"gefunden": gefunden, "neu": neu, "uebersprungen": uebersprungen,
|
||
"unklar": unklar, "seit": seit.isoformat(),
|
||
"ohne_zuordnung": ohne_zuordnung, "unklar_liste": unklar_liste}
|
||
|
||
|
||
def _nachbereiten(verbindung: imaplib.IMAP4, quelle: MailSource, uid: bytes) -> None:
|
||
"""Als gelesen markieren oder verschieben, damit nichts zweimal kommt."""
|
||
try:
|
||
if quelle.mark_seen:
|
||
verbindung.store(uid, "+FLAGS", "\\Seen")
|
||
if quelle.done_folder:
|
||
# MOVE gibt es nicht überall; COPY + DELETE ist der sichere Weg.
|
||
code, _ = verbindung.copy(uid, f'"{quelle.done_folder}"')
|
||
if code == "OK":
|
||
verbindung.store(uid, "+FLAGS", "\\Deleted")
|
||
except Exception as e: # Nachbereitung darf den Abruf nie kippen
|
||
log.warning("Mail konnte nicht nachbereitet werden: %s", e)
|
||
|
||
|
||
def abruf_mit_fehlerbehandlung(db: Session, quelle: MailSource) -> dict:
|
||
"""Wie ``hole_neue``, schreibt einen Fehler aber in die Quelle statt zu werfen."""
|
||
try:
|
||
return hole_neue(db, quelle)
|
||
except MailFehler as e:
|
||
quelle.last_error = str(e)[:500]
|
||
quelle.last_poll_at = datetime.now(timezone.utc)
|
||
db.commit()
|
||
return {"gefunden": 0, "neu": 0, "uebersprungen": 0, "unklar": 0,
|
||
"seit": None, "unklar_liste": [],
|
||
"ohne_zuordnung": [], "fehler": str(e)}
|