"""Az e-mail küldés (SMTP) beállításainak központi kezelése: betöltés, ellenőrzés, mentés, küldés. TÖRTÉNET / MIÉRT KÉSZÜLT (2026-10-04) Korábban az SMTP-adatok (szerver, port, felhasználó, jelszó) KIZÁRÓLAG a .env fájlból jöttek (config.py), és az alkalmazás indításakor egyszer töltődtek be - egy módosításhoz szerver- hozzáférés és újraindítás kellett. Az új igény: a rendszeradmin az alkalmazás felületéről (/rendszeradmin/email-beallitasok) tudja állítani ezeket. Ez a modul az egyetlen hely, ami az SMTP-adatokat tudja, minden más (email_utils.py, a rendszeradmin oldal) ezen keresztül éri el. AZ ADATOK FORRÁSA (prioritás) 1. ADATBÁZIS (rendszer_beallitasok tábla, models.RendszerBeallitas) - ha ott van elmentett "smtp_server". Az SMTP-jelszó titkosítva van tárolva (titkosito_utils.py). 2. .ENV TARTALÉK (config.SMTP_*) - ha az adatbázisban még nincs SMTP-szerver. Így a bevezetés után a levélküldés mindaddig a régi módon működik tovább, amíg a rendszeradmin el nem menti az első beállítást a felületen (vagy le nem futtatja a migracio_smtp_env_to_db.py-t). 3. Egyik sem: "szimulációs mód" - a levelek nem mennek ki, a szerver konzoljára íródnak ki (ez a korábbi viselkedés, fejlesztéshez hasznos). Az adatbázis és a .env értékei SOSEM keverednek: vagy az egyik, vagy a másik teljes készlete él. MIÉRT NINCS GYORSÍTÓTÁR (cache)? Az éles szerver gunicorn-nal 3 worker-folyamattal fut (server_tune/emelog.service). Egy folyamaton belüli cache-t a másik 2 folyamat nem látná, így egy mentés után akár percekig a régi beállítással mennének ki a levelek. Levélküldés ritka művelet (riasztás, napi riport), ezért inkább minden küldésnél újraolvassuk a DB-ből - egyszerű és mindig friss. MIT NE TEGYÉL? - A jelszót ne írd ki sehová (print, hiba, sablon). A beállítás-oldal sem jeleníti meg - csak azt, hogy "meg van adva". - Ne hívj közvetlenül smtplib-et máshonnan: használd a kuld() függvényt, hogy a időtúllépés és a TLS-ellenőrzés egységes maradjon. Részletes leírás: docs/EMAIL_RENDSZER.md """ import re import smtplib import ssl from dataclasses import dataclass from datetime import datetime, timezone from typing import Dict, List, Optional, Tuple import models from config import SMTP_SERVER, SMTP_PORT, SMTP_USER, SMTP_PASSWORD from database import SessionLocal from titkosito_utils import titkosit, visszafejt # --- A rendszer_beallitasok táblában használt kulcsok (EZEKET lehet csak elmenteni) --- KULCS_SZERVER = "smtp_server" KULCS_PORT = "smtp_port" KULCS_FELHASZNALO = "smtp_felhasznalo" KULCS_JELSZO = "smtp_jelszo" # titkosítva tárolt! KULCS_BIZTONSAG = "smtp_biztonsag" # "SSL" (465-ös port) vagy "STARTTLS" (587-es port) KULCS_FELADO_CIM = "smtp_felado_cim" # opcionális; ha üres, a felhasználónév a feladó KULCS_TLS_ELLENORZES = "smtp_tls_ellenorzes" # "1" = a szerver tanúsítványát ellenőrizzük, "0" = nem KULCS_ALKALMAZAS_URL = "alkalmazas_url" # a levelekben szereplő linkek alapja (pl. https://emelog.hu) BEALLITAS_KULCSOK = [ KULCS_SZERVER, KULCS_PORT, KULCS_FELHASZNALO, KULCS_JELSZO, KULCS_BIZTONSAG, KULCS_FELADO_CIM, KULCS_TLS_ELLENORZES, KULCS_ALKALMAZAS_URL, ] BIZTONSAG_ERTEKEK = ("SSL", "STARTTLS") # Ennyi másodpercig vár az SMTP-kapcsolat/válasz. FONTOS: korábban NEM volt időtúllépés, egy # elérhetetlen levelezőszerver a háttérfeladatot (és a gunicorn worker egy szálát) végtelenségig # blokkolhatta. SMTP_IDOTULLEPES_MP = 30 ALAP_ALKALMAZAS_URL = "http://localhost:8000" def _fejresz_tisztitas(ertek: str) -> str: """CR/LF-mentesítés (CRLF-fejléc-injekció ellen) - ugyanaz az elv, mint az email_utils.py-ban.""" return str(ertek).replace("\r", "").replace("\n", "") @dataclass class EmailBeallitasok: """Az aktuálisan érvényes (feloldott) e-mail beállítások egy példánya.""" szerver: str = "" port: int = 465 felhasznalo: str = "" jelszo: str = "" biztonsag: str = "SSL" felado_cim: str = "" tls_ellenorzes: bool = True alkalmazas_url: str = "" # "adatbazis" | "env" | "nincs" - honnan származik az SMTP-készlet (a beállító oldalon látszik) forras: str = "nincs" # False, ha az adatbázisban van jelszó, de nem fejthető vissza (pl. megváltozott a SECRET_KEY) jelszo_visszafejtheto: bool = True # Igaz, ha van (nem üres) jelszó - a beállító oldal ebből írja ki, hogy "meg van adva" jelszo_megadva: bool = False # Ha a betöltés közben hiba történt (pl. a tábla még nem létezik), ide kerül a rövid leírás betoltesi_hiba: str = "" @property def konfiguralt(self) -> bool: """Igaz, ha van annyi adat, hogy érdemes küldeni (a korábbi feltétellel azonos: szerver ÉS felhasználó megadva). Hamis -> szimulációs mód.""" return bool(self.szerver and self.felhasznalo) @property def felado_fejlec(self) -> str: """A 'From:' fejléc értéke. A feladó címe a külön megadott feladó-cím, ennek hiányában a belépési felhasználónév (a korábbi viselkedés). CR/LF-mentesítve.""" cim = _fejresz_tisztitas(self.felado_cim or self.felhasznalo) return f"EmeLOG Rendszer <{cim}>" @property def alkalmazas_url_vagy_alap(self) -> str: """A levelekbe kerülő linkek alapja, záró '/' nélkül; ha nincs beállítva, helyi cím.""" return (self.alkalmazas_url or ALAP_ALKALMAZAS_URL).rstrip("/") def _int_vagy(ertek, alap: int) -> int: try: return int(ertek) except (TypeError, ValueError): return alap def betolt(db=None) -> EmailBeallitasok: """Összeállítja az aktuális beállításokat (DB -> .env tartalék -> nincs). db: egy meglévő SQLAlchemy session (a rendszeradmin oldal ezt adja át). Ha None, a függvény SAJÁT, rövid életű sessiont nyit - ez kell a háttérfeladatoknak és az ütemezett jobnak (azoknak nincs kérés-kontextusukhoz kötött session-jük). A függvény SOSEM dob kivételt: ha az adatbázis nem olvasható (pl. a tábla még nem létezik egy frissen frissített, de még nem újraindított szerveren), a .env tartalékra esik vissza. """ sajat_session = db is None ertekek: Dict[str, Tuple[Optional[str], bool]] = {} hiba = "" if sajat_session: db = SessionLocal() try: sorok = db.query(models.RendszerBeallitas).filter( models.RendszerBeallitas.kulcs.in_(BEALLITAS_KULCSOK) ).all() # Egyszerű (nem ORM) értékekké alakítjuk, hogy a session lezárása után is használhatók legyenek ertekek = {s.kulcs: (s.ertek, bool(s.titkositott)) for s in sorok} except Exception as e: hiba = f"{type(e).__name__}: {e}" print(f"❌ Hiba az e-mail beállítások DB-ből olvasásakor (.env tartalék használatban): {hiba}") ertekek = {} finally: if sajat_session: db.close() def ertek(kulcs: str) -> str: v = ertekek.get(kulcs) return (v[0] or "").strip() if v else "" b = EmailBeallitasok(alkalmazas_url=ertek(KULCS_ALKALMAZAS_URL), betoltesi_hiba=hiba) if ertek(KULCS_SZERVER): # --- ADATBÁZIS-ALAPÚ beállítás --- b.forras = "adatbazis" b.szerver = ertek(KULCS_SZERVER) b.port = _int_vagy(ertek(KULCS_PORT), 465) b.felhasznalo = ertek(KULCS_FELHASZNALO) b.biztonsag = ertek(KULCS_BIZTONSAG) if ertek(KULCS_BIZTONSAG) in BIZTONSAG_ERTEKEK else "SSL" b.felado_cim = ertek(KULCS_FELADO_CIM) # Alapértelmezés: ellenőrzünk (csak akkor kapcsol ki, ha kifejezetten "0" van elmentve) b.tls_ellenorzes = ertek(KULCS_TLS_ELLENORZES) != "0" jelszo_sor = ertekek.get(KULCS_JELSZO) if jelszo_sor and jelszo_sor[0]: b.jelszo_megadva = True if jelszo_sor[1]: feloldott = visszafejt(jelszo_sor[0]) if feloldott is None: b.jelszo = "" b.jelszo_visszafejtheto = False else: b.jelszo = feloldott else: b.jelszo = jelszo_sor[0] elif SMTP_SERVER: # --- .ENV TARTALÉK (a régi viselkedés bájtra pontosan: SSL, tanúsítvány-ellenőrzés nélkül) --- b.forras = "env" b.szerver = SMTP_SERVER b.port = SMTP_PORT b.felhasznalo = SMTP_USER b.jelszo = SMTP_PASSWORD b.jelszo_megadva = bool(SMTP_PASSWORD) b.biztonsag = "SSL" b.tls_ellenorzes = False # különben: forras == "nincs" (szimulációs mód) return b # --------------------------------------------------------------------------------------------- # ELLENŐRZÉS (validáció) - a rendszeradmin oldal használja mentés előtt # --------------------------------------------------------------------------------------------- _SZERVER_MINTA = re.compile(r"^[A-Za-z0-9]([A-Za-z0-9.\-]{0,253})$") _EMAIL_MINTA = re.compile(r"^[^@\s<>\"'\\]+@[^@\s<>\"'\\]+\.[^@\s<>\"'\\]+$") _URL_MINTA = re.compile(r"^https?://[^\s<>\"']+$") _VEZERLOKARAKTER = re.compile(r"[\x00-\x1f\x7f]") def ellenoriz(nyers: Dict[str, str]) -> Tuple[Dict[str, str], List[str]]: """Megtisztítja és ellenőrzi a beküldött űrlap-értékeket. Visszatérés: (tiszta_ertekek, hibauzenetek). Ha a hibalista nem üres, NEM szabad menteni. A bemeneti kulcsok: szerver, port, felhasznalo, jelszo, biztonsag, felado_cim, tls_ellenorzes ("1"/"0"), alkalmazas_url. A 'jelszo' üres szövege = "ne módosítsd". """ hibak: List[str] = [] t: Dict[str, str] = {} szerver = (nyers.get("szerver") or "").strip() if szerver and not _SZERVER_MINTA.match(szerver): hibak.append("Az SMTP-szerver neve érvénytelen (csak betű, szám, pont és kötőjel lehet benne, szóköz nélkül).") t["szerver"] = szerver port_szoveg = (nyers.get("port") or "").strip() try: port = int(port_szoveg) if port_szoveg else 465 if not (1 <= port <= 65535): raise ValueError except ValueError: hibak.append("A port 1 és 65535 közötti egész szám kell legyen.") port = 465 t["port"] = str(port) felhasznalo = (nyers.get("felhasznalo") or "").strip() if len(felhasznalo) > 255 or _VEZERLOKARAKTER.search(felhasznalo): hibak.append("A felhasználónév túl hosszú vagy tiltott (vezérlő) karaktert tartalmaz.") t["felhasznalo"] = felhasznalo jelszo = nyers.get("jelszo") or "" if len(jelszo) > 500 or _VEZERLOKARAKTER.search(jelszo): hibak.append("A jelszó túl hosszú vagy tiltott (vezérlő, pl. sortörés) karaktert tartalmaz.") t["jelszo"] = jelszo biztonsag = (nyers.get("biztonsag") or "SSL").strip().upper() if biztonsag not in BIZTONSAG_ERTEKEK: hibak.append("A titkosítás típusa csak SSL vagy STARTTLS lehet.") biztonsag = "SSL" t["biztonsag"] = biztonsag felado = (nyers.get("felado_cim") or "").strip() if felado and (len(felado) > 255 or not _EMAIL_MINTA.match(felado)): hibak.append("A feladó-cím nem érvényes e-mail cím.") t["felado_cim"] = felado t["tls_ellenorzes"] = "1" if (nyers.get("tls_ellenorzes") or "1") == "1" else "0" url = (nyers.get("alkalmazas_url") or "").strip().rstrip("/") if url and (len(url) > 255 or not _URL_MINTA.match(url)): hibak.append("Az alkalmazás URL-je http:// vagy https:// kezdetű, szóközmentes cím kell legyen (pl. https://emelog.hu).") t["alkalmazas_url"] = url # Ha szerver meg van adva, a felhasználónév is kell (különben a küldés nem konfigurált) if szerver and not felhasznalo: hibak.append("Ha megadsz SMTP-szervert, a felhasználónevet is meg kell adni.") return t, hibak # --------------------------------------------------------------------------------------------- # MENTÉS # --------------------------------------------------------------------------------------------- def _upsert(db, kulcs: str, ertek: str, titkositott: bool, user_id: Optional[int]) -> bool: """Egy kulcs beszúrása/frissítése. Igazzal tér vissza, ha tényleg változott az érték.""" sor = db.query(models.RendszerBeallitas).filter(models.RendszerBeallitas.kulcs == kulcs).first() if sor is None: db.add(models.RendszerBeallitas( kulcs=kulcs, ertek=ertek, titkositott=titkositott, modositva_at=datetime.now(timezone.utc), modosito_user_id=user_id, )) return True if (sor.ertek or "") == ertek and bool(sor.titkositott) == titkositott: return False sor.ertek = ertek sor.titkositott = titkositott sor.modositva_at = datetime.now(timezone.utc) sor.modosito_user_id = user_id return True # A módosított mezők emberi neve a naplóbejegyzéshez (a jelszó értékét SOHA nem naplózzuk) _MEZO_CIMKEK = { KULCS_SZERVER: "SMTP-szerver", KULCS_PORT: "port", KULCS_FELHASZNALO: "felhasználónév", KULCS_JELSZO: "jelszó", KULCS_BIZTONSAG: "titkosítás típusa", KULCS_FELADO_CIM: "feladó-cím", KULCS_TLS_ELLENORZES: "TLS-tanúsítvány ellenőrzése", KULCS_ALKALMAZAS_URL: "alkalmazás URL", } def ment(db, tiszta: Dict[str, str], user_id: Optional[int]) -> List[str]: """Elmenti az ellenőrzött értékeket. NEM commitol - a hívó (a router) az AuditLog-bejegyzéssel együtt, EGY tranzakcióban commitolja. A jelszó csak akkor íródik felül, ha a 'jelszo' mező nem üres (üres = változatlan). Visszatérés: a ténylegesen módosult mezők emberi nevei (naplózáshoz). """ valtozott: List[str] = [] def mentes(kulcs, ertek, titkositott=False): if _upsert(db, kulcs, ertek, titkositott, user_id): valtozott.append(_MEZO_CIMKEK[kulcs]) mentes(KULCS_SZERVER, tiszta["szerver"]) mentes(KULCS_PORT, tiszta["port"]) mentes(KULCS_FELHASZNALO, tiszta["felhasznalo"]) mentes(KULCS_BIZTONSAG, tiszta["biztonsag"]) mentes(KULCS_FELADO_CIM, tiszta["felado_cim"]) mentes(KULCS_TLS_ELLENORZES, tiszta["tls_ellenorzes"]) mentes(KULCS_ALKALMAZAS_URL, tiszta["alkalmazas_url"]) if tiszta.get("jelszo"): # A titkosítás minden mentésnél új "sót" (IV-t) használ, ezért az _upsert mindig # változásnak látja - ez itt helyes, mert csak akkor jutunk ide, ha új jelszót adtak meg. mentes(KULCS_JELSZO, titkosit(tiszta["jelszo"]), titkositott=True) return valtozott # --------------------------------------------------------------------------------------------- # KÜLDÉS # --------------------------------------------------------------------------------------------- def kuld(msg, beall: Optional[EmailBeallitasok] = None) -> None: """Elküld egy kész MIME-üzenetet a beállított SMTP-szerveren. Hiba esetén KIVÉTELT dob (a hívó dönti el, mit kezd vele: a riasztás-küldők naplózzák, a teszt-levél gomb megmutatja). Ha nincs konfigurálva SMTP (szimulációs mód), RuntimeError-t dob - a hívóknak ezt a küldés ELŐTT, a `beall.konfiguralt` ellenőrzésével kell kezelniük. """ if beall is None: beall = betolt() if not beall.konfiguralt: raise RuntimeError("Nincs SMTP-szerver beállítva (szimulációs mód).") tls_kontextus = ssl.create_default_context() if not beall.tls_ellenorzes: # A korábbi viselkedés (az smtplib alapértelmezése): nincs tanúsítvány-ellenőrzés. Csak a # .env-tartalék és az admin által kifejezetten kikapcsolt eset használja - ilyenkor egy # hálózati közbeékelődő támadó (MITM) elvileg lehallgathatná a jelszót, ezért a # rendszeradmin oldal figyelmeztet erre. tls_kontextus.check_hostname = False tls_kontextus.verify_mode = ssl.CERT_NONE if beall.biztonsag == "STARTTLS": szerver = smtplib.SMTP(beall.szerver, beall.port, timeout=SMTP_IDOTULLEPES_MP) try: szerver.ehlo() szerver.starttls(context=tls_kontextus) szerver.ehlo() szerver.login(beall.felhasznalo, beall.jelszo) szerver.send_message(msg) finally: try: szerver.quit() except Exception: pass else: szerver = smtplib.SMTP_SSL(beall.szerver, beall.port, timeout=SMTP_IDOTULLEPES_MP, context=tls_kontextus) try: szerver.login(beall.felhasznalo, beall.jelszo) szerver.send_message(msg) finally: try: szerver.quit() except Exception: pass