"""Bejelentkezés-védelem (brute-force ellen): sikertelen kísérletek számlálása az ADATBÁZISBAN, tiltások, a rendszeradmin értesítése és a tiltás feloldása. (2026-10-04, D szelet) Részletes leírás Miklósnak és a jövőbeli fejlesztőknek: docs/LOGIN_VEDELEM.md MI VÁLTOZOTT A RÉGIHEZ KÉPEST? Régen (security.py) a próbálkozásokat a folyamat memóriája számolta: 3 gunicorn workernél workerenként külön (a valódi határ így akár 3x annyi volt), újraindításkor nullázódott, és senki nem tudott róla, ha valakit kizártak. Most: - a számlálás a `login_kiserletek` táblában van (minden worker ugyanazt látja), - a határ elérésekor TILTÁS keletkezik (`login_tiltasok`), nem csak "pár percig nem enged", - valós fiók tiltásáról a rendszeradmin e-mailt kap, - ismétlődő tiltásnál a tiltás VÉGLEGES lesz, amíg a rendszeradmin (csak ő!) fel nem oldja. SZABÁLYOK Határ (5 perces ablakban): 8 sikertelen kísérlet egy fiókra (e-mailre) VAGY 20 egy IP-ről. Tiltás hossza: 15 perc; ha ugyanarra az azonosítóra 24 órán belül ismét tiltás kell, 60 perc; a HARMADIK 24 órán belüli tiltás VÉGLEGES - de csak VALÓS, NEM rendszeradmin fiók esetén. A rendszeradmin fiókját SOHA nem lehet véglegesen kizárni (különben egy támadó örökre kizárhatná az egyetlen embert, aki a tiltásokat oldhatja) - őt csak időzítve tiltjuk. IP-tiltás sosem végleges (egy közös céges NAT-ot nem akarunk örökre kizárni). "VALÓS FIÓK" (a felhasználó kifejezett kérése: "csak valós felhasználók esetére") - fiók (e-mail) tiltásnál: az e-mail egy LÉTEZŐ felhasználóé (kis/nagybetű nélkül); - IP-tiltásnál: az adott IP-ről az ablakban legalább egy LÉTEZŐ fiókot próbáltak. Csak valós tiltásról megy értesítő levél és csak valós fiók kaphat végleges tiltást. Ismeretlen címekre (botok, elírás) időzített tiltás van, értesítés nélkül. AZ ENUMERÁCIÓ ELLENI VÉDELEM A /login válasza minden esetben ugyanaz (lásd routers/auth.py): a tiltott és a nem tiltott válasza ugyanaz a szöveg, és a "nem létező fiók" is ugyanúgy viselkedik. A felhasználó szempontjából nem derül ki, hogy a cím létezik-e. (Egyetlen finom különbség: a nem létező címek tiltása 15 perc után mindig feloldódik, a valósé véglegessé válhat - ezt hosszan, sok tiltás után lehetne megfigyelni; ezt tudatosan vállaljuk, lásd docs/LOGIN_VEDELEM.md.) TRANZAKCIÓ: a függvények a hívó `db` session-jét használják és maguk commitolnak (a /login kérés végén nincs más módosítás). Hiba esetén a bejelentkezést NEM állítják meg: a számlálás vagy az értesítés elromlása nem akadályozhatja a bejelentkezést (a hiba a konzolra kerül). """ from datetime import datetime, timedelta, timezone from typing import List, Optional from sqlalchemy import func from sqlalchemy.exc import IntegrityError import models import email_beallitasok import email_sor from database import SessionLocal from templates_env import HELYI_TZ # --- Beállítások (a régi security.py-val azonos határok) ------------------------------------- ABLAK_MASODPERC = 300 # ennyi időre visszamenőleg számoljuk a sikertelen kísérleteket MAX_KISERLET_IP = 20 MAX_KISERLET_EMAIL = 8 ELSO_TILTAS_PERC = 15 # első tiltás hossza ISMETELT_TILTAS_PERC = 60 # ismételt tiltás (24 órán belül) hossza VEGLEGES_ENNYI_TILTAS_UTAN = 3 # a 24 órán belüli HARMADIK tiltás végleges (valós, nem rendszeradmin fióknál) ISMETLES_ABLAK_ORA = 24 ERTESITES_MAX_ORANKENT = 10 # ennyi egyedi értesítő megy óránként; felette napi összefoglaló KEZELT_HOSSZ = 190 # az e-mail / azonosító oszlopok hossza (utf8mb4 index-határ miatt) # Takarítás KISERLET_MEGTARTAS_NAP = 1 TILTAS_MEGTARTAS_NAP = 90 ISMERETLEN_TILTAS_MEGTARTAS_NAP = 7 GENERIKUS_HIBA = "Hibás e-mail cím vagy jelszó!" TILTOTT_UZENET = ("Túl sok sikertelen bejelentkezési kísérlet történt erről a címről/fiókról. " "Kérjük, várj egy kicsit, mielőtt újra próbálkozol! Ha ez többször előfordul, " "fordulj a rendszergazdához.") def _utc_most() -> datetime: """Tz nélküli UTC pillanat (a DB-ben így tároljuk az időpontokat).""" return datetime.now(timezone.utc).replace(tzinfo=None) def _email_kulcs(email: Optional[str]) -> str: return (email or "").strip().lower()[:KEZELT_HOSSZ] def _ip_kulcs(ip: Optional[str]) -> str: return (ip or "ismeretlen").strip()[:64] def _rendszeradmin_e(user) -> bool: # Ugyanaz a szabály, mint security.is_rendszeradminisztrator (nem importáljuk: körkörös import lenne) return user is not None and user.role == "Admin" and user.company_id == 1 def _felhasznalo_email_szerint(db, email_kisbetus: str): if not email_kisbetus: return None return db.query(models.User).filter(func.lower(models.User.email) == email_kisbetus).first() # ============================================================================================= # 1. ELLENŐRZÉS ÉS RÖGZÍTÉS (a /login hívja) # ============================================================================================= def _aktiv_tiltasok_lekerdezes(db, most: datetime): T = models.LoginTiltas return db.query(T).filter( T.feloldva_at.is_(None), (T.lejar_at.is_(None)) | (T.lejar_at > most), ) def tiltva_e(db, ip: str, email: str) -> bool: """Igaz, ha az IP-re VAGY az e-mailre aktív tiltás van (nem feloldott, és nem járt le).""" T = models.LoginTiltas ip_k, email_k = _ip_kulcs(ip), _email_kulcs(email) q = _aktiv_tiltasok_lekerdezes(db, _utc_most()).filter( ((T.tipus == "ip") & (T.azonosito == ip_k)) | ((T.tipus == "email") & (T.azonosito == email_k)) ) return db.query(q.exists()).scalar() is True def sikertelen_rogzitese(db, ip: str, email: str, user=None) -> Optional["models.LoginTiltas"]: """Egy sikertelen kísérlet rögzítése; ha ezzel elértük a határt, tiltást hoz létre. Visszatérés: az újonnan létrehozott tiltás (vagy None). SOSEM dob kivételt.""" try: # FRISS ADATBÁZIS-PILLANATKÉP: a /login az elején már olvasott (tiltva_e, felhasználó-keresés), és # MariaDB-n (REPEATABLE READ) a tranzakció az első olvasáskor "befagyasztja" a látott adatokat - # a közben (a bcrypt ~100-300 ms alatt) más worker által commitolt kísérleteket nem látnánk, # és egy párhuzamos roham minden kérése alulszámolná a próbálkozásokat (a tiltás elmaradna). # A rollback itt semmit nem dob el (nincs függő módosítás), csak új pillanatképet nyit. db.rollback() most = _utc_most() ip_k, email_k = _ip_kulcs(ip), _email_kulcs(email) db.add(models.LoginKiserlet(ip=ip_k, email=email_k, idopont=most)) db.flush() ablak_kezdet = most - timedelta(seconds=ABLAK_MASODPERC) K = models.LoginKiserlet ip_db = db.query(func.count(K.id)).filter(K.ip == ip_k, K.idopont >= ablak_kezdet).scalar() or 0 email_db = db.query(func.count(K.id)).filter(K.email == email_k, K.idopont >= ablak_kezdet).scalar() or 0 uj = None if email_k and email_db >= MAX_KISERLET_EMAIL and not _van_aktiv_tiltas(db, "email", email_k, most): uj = _tiltas_letrehoz(db, "email", email_k, ip_k, email_db, user, most) if ip_db >= MAX_KISERLET_IP and not _van_aktiv_tiltas(db, "ip", ip_k, most): ip_tiltas = _tiltas_letrehoz(db, "ip", ip_k, ip_k, ip_db, None, most, ablak_kezdet=ablak_kezdet) uj = uj or ip_tiltas db.commit() return uj except Exception as e: db.rollback() print(f"❌ Login-védelem: a sikertelen kísérlet rögzítése nem sikerült: {type(e).__name__}: {e}") return None def sikeres_torlese(db, ip: str, email: str) -> None: """Sikeres bejelentkezés után az adott FIÓK sikertelen kísérleteinek törlése. Szándékos változás a régihez képest: az IP más fiókokra vonatkozó kísérleteit NEM töröljük (régen egy sikeres bejelentkezés az egész IP számlálóját nullázta - egy támadó a saját, érvényes fiókjával időnként "kinullázhatta" volna a saját IP-je számlálóját). SOSEM dob kivételt.""" try: db.query(models.LoginKiserlet).filter(models.LoginKiserlet.email == _email_kulcs(email)).delete(synchronize_session=False) db.commit() except Exception as e: db.rollback() print(f"❌ Login-védelem: a kísérletek törlése nem sikerült: {type(e).__name__}: {e}") def _van_aktiv_tiltas(db, tipus: str, azonosito: str, most: datetime) -> bool: T = models.LoginTiltas q = _aktiv_tiltasok_lekerdezes(db, most).filter(T.tipus == tipus, T.azonosito == azonosito) return db.query(q.exists()).scalar() is True # ============================================================================================= # 2. TILTÁS LÉTREHOZÁSA ÉS ÉRTESÍTÉS # ============================================================================================= def _valos_fiokok_az_ip_rol(db, ip_k: str, ablak_kezdet: datetime) -> List[str]: """Az IP-ről az ablakban próbált e-mail címek közül a LÉTEZŐ felhasználók címei (legfeljebb 10).""" K = models.LoginKiserlet cimek = [r[0] for r in db.query(K.email).filter(K.ip == ip_k, K.idopont >= ablak_kezdet).distinct().limit(50).all() if r[0]] if not cimek: return [] letezok = db.query(func.lower(models.User.email)).filter(func.lower(models.User.email).in_(cimek)).all() return sorted({r[0] for r in letezok})[:10] def _tiltas_letrehoz(db, tipus: str, azonosito: str, ip_k: str, probalkozasok: int, user, most: datetime, ablak_kezdet: Optional[datetime] = None): T = models.LoginTiltas erintett: List[str] = [] valos = False felhasznalo = None if tipus == "email": felhasznalo = user or _felhasznalo_email_szerint(db, azonosito) valos = felhasznalo is not None else: erintett = _valos_fiokok_az_ip_rol(db, ip_k, ablak_kezdet or (most - timedelta(seconds=ABLAK_MASODPERC))) valos = bool(erintett) # Eszkaláció: hány tiltás volt ugyanerre az azonosítóra az elmúlt 24 órában - a rendszeradmin # LEGUTÓBBI kézi feloldása óta (egy feloldás "tiszta lapot" ad: az éppen feloldott, jogos # felhasználót az első elgépelés után ne zárjuk ki azonnal véglegesen). ablak = most - timedelta(hours=ISMETLES_ABLAK_ORA) utolso_feloldas = db.query(func.max(T.feloldva_at)).filter( T.tipus == tipus, T.azonosito == azonosito, T.felolto_user_id.isnot(None)).scalar() kezdet = max(ablak, utolso_feloldas) if utolso_feloldas else ablak korabbi = db.query(func.count(T.id)).filter( T.tipus == tipus, T.azonosito == azonosito, T.letrehozva_at >= kezdet).scalar() or 0 vegleges = ( tipus == "email" and valos and not _rendszeradmin_e(felhasznalo) and (korabbi + 1) >= VEGLEGES_ENNYI_TILTAS_UTAN ) perc = ELSO_TILTAS_PERC if korabbi == 0 else ISMETELT_TILTAS_PERC lejar = None if vegleges else most + timedelta(minutes=perc) # Az "aktiv_kulcs" egyedi őr felszabadítása a már LEJÁRT / feloldott tiltásoknál (különben egy lejárt # tiltás sora örökre blokkolná az azonosító új tiltását) kulcs = f"{tipus}:{azonosito}" db.query(T).filter( T.aktiv_kulcs == kulcs, T.feloldva_at.isnot(None) | (T.lejar_at.isnot(None) & (T.lejar_at <= most)) ).update({"aktiv_kulcs": None}, synchronize_session=False) sor = T( tipus=tipus, azonosito=azonosito, letrehozva_at=most, lejar_at=lejar, vegleges=vegleges, probalkozasok=probalkozasok, ip_cim=ip_k, user_id=felhasznalo.id if felhasznalo else None, company_id=felhasznalo.company_id if felhasznalo else None, valos=valos, erintett=", ".join(erintett) if erintett else None, aktiv_kulcs=kulcs, ) try: with db.begin_nested(): db.add(sor) except IntegrityError: # Egy másik worker ugyanebben a pillanatban már létrehozta ugyanezt a tiltást - nem hozunk másodikat # (különben egy roham 2-nek számítana az eszkalációban, és a feloldás után is maradna egy "iker" tiltás). return None if valos: # A naplózás és az értesítés hibája NEM veheti el a tiltást: külön SAVEPOINT-ban fut try: with db.begin_nested(): # Jogosultsági napló (user_id nélkül: nem a rendszeradmin tette, a rendszer) db.add(models.AuditLog( company_id=sor.company_id, field_name="Bejelentkezési tiltás", old_value="-", new_value=(f"{'Fiók' if tipus == 'email' else 'IP'}: {azonosito}; " f"{'VÉGLEGES (csak a rendszeradmin oldhatja)' if vegleges else f'{perc} perc'}; " f"{probalkozasok} sikertelen kísérlet"), )) _ertesites(db, sor, felhasznalo, most) db.flush() except Exception as e: print(f"❌ Login-védelem: a tiltás naplózása/értesítése nem sikerült (a tiltás él): {type(e).__name__}: {e}") sor.ertesites = "hiba" else: sor.ertesites = "nincs_valos" return sor def _rendszeradmin_cimek(db) -> List[str]: sorok = db.query(models.User.email).filter( models.User.role == "Admin", models.User.company_id == 1, models.User.aktiv.is_(True), models.User.aktivalva.isnot(False), # a még meg nem erősített cím nem kap riasztást models.User.email.isnot(None)).all() return [r[0] for r in sorok if r[0]] def rendszeradmin_cimek(db) -> List[str]: """Az aktív rendszeradminisztrátorok e-mail címei (más modulok értesítéseihez, pl. a szimultán belépés jelzéséhez).""" return _rendszeradmin_cimek(db) def _ertesites(db, sor, felhasznalo, most: datetime) -> None: """Értesítő levél a rendszeradminoknak (sürgős prioritással) - csak valós tiltásról hívjuk. Árvíz-védelem: óránként legfeljebb ERTESITES_MAX_ORANKENT egyedi levél; felette naponta EGY összefoglaló ("sok tiltás történt") - így egy támadás nem öntheti el a postafiókot (és nem használja el a levél-korlátot sem). A VÉGLEGES tiltásról a korlát ellenére is megy levél: azt a rendszeradminnak tudnia KELL (és számuk eleve kicsi: fiókonként legalább 3 előzmény kell hozzá).""" import html _e = lambda x: html.escape("" if x is None else str(x), quote=True) # noqa: E731 cimek = _rendszeradmin_cimek(db) if not cimek: sor.ertesites = "nincs_cimzett" return T = models.LoginTiltas elmult_ora = db.query(func.count(T.id)).filter( T.ertesites == "elkuldve", T.ertesitve_at >= most - timedelta(hours=1)).scalar() or 0 beall = email_beallitasok.betolt(db) link = beall.alkalmazas_url_vagy_alap.rstrip("/") + "/rendszeradmin/kitiltasok" if elmult_ora >= ERTESITES_MAX_ORANKENT and not sor.vegleges: sor.ertesites = "korlat" ma = datetime.now(HELYI_TZ).date() if email_sor.kikuldheto_e_es_rogzit(db, f"bann:aradat:{ma.isoformat()}", None, 1, ma): torzs = (f"

Ma több mint {ERTESITES_MAX_ORANKENT} bejelentkezési tiltás történt óránként, ezért a " f"további tiltásokról nem küldünk külön levelet.

" f"

A tiltások listája és feloldásuk

") email_sor.sorba_allit(cimek, "Sok bejelentkezési tiltás történt", torzs, kategoria="rendszer", prioritas=email_sor.PRIORITAS_SURGOS, db=db) return if sor.tipus == "email": ki = f"a(z) {_e(sor.azonosito)} fiók" if felhasznalo is not None: ki += f" ({_e(felhasznalo.username)}" ceg = db.get(models.Company, felhasznalo.company_id) if felhasznalo.company_id else None if ceg is not None: ki += f", {_e(ceg.nev)}" ki += ")" targy = f"Bejelentkezési tiltás: {sor.azonosito}" else: ki = f"a(z) {_e(sor.azonosito)} IP-cím" targy = "Bejelentkezési tiltás: IP-cím" if sor.vegleges: mennyi = ("VÉGLEGES a tiltás: addig tart, amíg a rendszeradminisztrátor fel nem oldja " "(csak Te oldhatod fel).") else: mennyi = f"A tiltás {_e(helyi_ido_szoveg(sor.lejar_at))}-ig tart, utána magától megszűnik." erintett = "" if sor.erintett: erintett = f"

Érintett (létező) fiókok: {_e(sor.erintett)}

" torzs = ( f"

Kitiltottuk {ki}, mert {_e(sor.probalkozasok)} sikertelen bejelentkezési kísérlet történt " f"5 percen belül (IP-cím: {_e(sor.ip_cim)}).

" f"

{mennyi}

{erintett}" f"

Ha a felhasználó jogosult, a tiltást feloldhatod: Tiltások kezelése.

" f"

Ha nem a felhasználó próbálkozott, ne oldd fel: " f"a tiltás védi a fiókot a jelszófeltörési kísérlettől.

" ) db_szam = email_sor.sorba_allit(cimek, targy, torzs, kategoria="rendszer", prioritas=email_sor.PRIORITAS_SURGOS, db=db) if db_szam: sor.ertesites = "elkuldve" sor.ertesitve_at = most else: sor.ertesites = "nincs_cimzett" def helyi_ido_szoveg(dt: Optional[datetime]) -> str: """Tz nélküli UTC időpont helyi (budapesti) időként, szövegesen.""" if dt is None: return "-" return dt.replace(tzinfo=timezone.utc).astimezone(HELYI_TZ).strftime("%Y-%m-%d %H:%M") # ============================================================================================= # 3. A RENDSZERADMIN FELÜLETÉHEZ # ============================================================================================= def aktiv_tiltasok(db): """Az aktív tiltások (valós és ismeretlen is), legújabb elöl.""" T = models.LoginTiltas return _aktiv_tiltasok_lekerdezes(db, _utc_most()).order_by(T.letrehozva_at.desc()).all() def legutobbi_tiltasok(db, limit: int = 50): """A legutóbbi tiltások története - csak a VALÓS fiókokat érintők. Visszatérés: (lista, az elmúlt 24 órában keletkezett ismeretlen címes tiltások száma).""" T = models.LoginTiltas lista = db.query(T).filter(T.valos.is_(True)).order_by(T.letrehozva_at.desc()).limit(limit).all() ismeretlen = db.query(func.count(T.id)).filter( T.valos.is_(False), T.letrehozva_at >= _utc_most() - timedelta(hours=24)).scalar() or 0 return lista, ismeretlen def _feloldas(db, sor, most: datetime, felolto_id: Optional[int], felolto_szoveg: str) -> None: """A tiltás tényleges feloldása (közös a webes és a parancssoros úton). NEM commitol.""" sor.feloldva_at = most sor.felolto_user_id = felolto_id sor.aktiv_kulcs = None # A kísérletek törlése: különben a (még az ablakban lévő) régi kísérletek azonnal újra tiltanának K = models.LoginKiserlet if sor.tipus == "email": db.query(K).filter(K.email == sor.azonosito).delete(synchronize_session=False) else: db.query(K).filter(K.ip == sor.azonosito).delete(synchronize_session=False) db.add(models.AuditLog( company_id=sor.company_id if sor.company_id is not None else 1, user_id=felolto_id, field_name="Bejelentkezési tiltás feloldva", old_value=f"{'Fiók' if sor.tipus == 'email' else 'IP'}: {sor.azonosito}" f"{' (végleges)' if sor.vegleges else ''}", new_value=f"Feloldotta: {felolto_szoveg}", )) def felold(db, tiltas_id: int, rendszeradmin) -> bool: """A tiltás kézi feloldása. FIGYELEM: a jogosultságot (csak rendszeradmin) a HÍVÓ ellenőrzi (routers/rendszeradmin_kitiltasok.py) - itt védelmi mélységként még egyszer megnézzük. Visszatérés: igaz, ha feloldottunk egy aktív tiltást.""" if not _rendszeradmin_e(rendszeradmin): return False most = _utc_most() T = models.LoginTiltas sor = _aktiv_tiltasok_lekerdezes(db, most).filter(T.id == tiltas_id).first() if sor is None: return False _feloldas(db, sor, most, rendszeradmin.id, rendszeradmin.username) db.commit() return True def felold_parancssorbol(db, azonosito: Optional[str] = None) -> List["models.LoginTiltas"]: """VÉSZFELOLDÁS a szerverről (login_felold.py) arra az esetre, ha MINDEN rendszeradmint kitiltottak, és így senki nem tud belépni a felületre. Aki a szerveren parancsot futtathat, az eleve a rendszer ura, ezért itt nincs külön jogosultság-ellenőrzés. `azonosito`: egy e-mail vagy IP; None = az összes aktív tiltás. Visszatérés: a feloldott tiltások.""" most = _utc_most() T = models.LoginTiltas q = _aktiv_tiltasok_lekerdezes(db, most) if azonosito: q = q.filter(T.azonosito == _email_kulcs(azonosito)) if "@" in azonosito else q.filter(T.azonosito == _ip_kulcs(azonosito)) sorok = q.all() for sor in sorok: _feloldas(db, sor, most, None, "szerver-parancssor (vészfeloldás)") db.commit() return sorok # ============================================================================================= # 4. TAKARÍTÁS (naponta egyszer, l. main.py) # ============================================================================================= def takaritas() -> dict: """Régi kísérletek és tiltások törlése. SOSEM dob kivételt.""" stat = {"kiserletek": 0, "tiltasok": 0} db = SessionLocal() try: most = _utc_most() stat["kiserletek"] = db.query(models.LoginKiserlet).filter( models.LoginKiserlet.idopont < most - timedelta(days=KISERLET_MEGTARTAS_NAP) ).delete(synchronize_session=False) T = models.LoginTiltas # Véget ért (lejárt vagy feloldott) valós tiltások 90 nap után; az ismeretlenek 7 nap után. # Az AKTÍV (köztük a végleges) tiltást soha nem töröljük. vege = (T.feloldva_at.isnot(None)) | ((T.lejar_at.isnot(None)) & (T.lejar_at < most)) stat["tiltasok"] += db.query(T).filter( vege, T.valos.is_(True), T.letrehozva_at < most - timedelta(days=TILTAS_MEGTARTAS_NAP) ).delete(synchronize_session=False) stat["tiltasok"] += db.query(T).filter( vege, T.valos.is_(False), T.letrehozva_at < most - timedelta(days=ISMERETLEN_TILTAS_MEGTARTAS_NAP) ).delete(synchronize_session=False) db.commit() except Exception as e: db.rollback() print(f"❌ Hiba a login-védelem takarításakor: {type(e).__name__}: {e}") finally: db.close() return stat