"""Kétlépcsős hitelesítés: a belépés 2. lépése - a 6 jegyű kód bekérése (2. fázis), és az első beállítás (3. fázis). FOLYAMAT (kész beállítás, 2. fázis) 1. POST /login: helyes jelszó + kötelező 2FA + kész beállítás -> átmeneti (mfa_pending, cel="kod") süti, átirányítás ide 2. GET /mfa : kód-bevitel oldal (csak érvényes átmeneti sütivel) 3. POST /mfa : a kód ellenőrzése. Siker -> teljes munkamenet-süti (`mfa` jelzéssel), a számlálók nullázása, naplózás. Hiba -> a sikertelen kísérlet számlálása a login_vedelem-ben (ugyanaz a tiltás, mint a jelszó-találgatásnál: túl sok próbálkozás után tiltás + a rendszeradmin értesítése). 4. POST /mfa/megse: az átmeneti süti törlése, vissza a bejelentkezéshez. FOLYAMAT (első beállítás, 3. fázis) - ha a felhasználóra kötelező a 2FA, de még nincs megerősített beállítása 1. POST /login: helyes jelszó -> átmeneti (cel="beallitas", 15 perc) süti, átirányítás a /mfa/beallitas-ra 2. GET /mfa/beallitas : (v28-tól) csak a BEVEZETŐ + "QR-kód létrehozása" gomb - NEM ír (egy GET-nek ismételhetőnek kell lennie) POST /mfa/beallitas/inditas : új TOTP-titok (a korábbi meg nem erősített eldobva), a válasz közvetlenül a QR-kód + kézi kulcs oldal 3. POST /mfa/beallitas : a hitelesítő alkalmazás első kódja. Siker -> a beállítás KÉSZ, a felhasználó be van lépve (a kód ismerete bizonyítja, hogy a titok a telefonján van), naplózás, értesítő e-mail A MEGERŐSÍTETT beállítást ez az oldal NEM írja felül (csak a "beallitando" állapotú felhasználó érheti el) - különben egy ellopott jelszó kicserélhetné a második faktort. Elveszett telefon: rendszeradmin-reset (mfa_felold.py; 5. fázis: felület). BIZTONSÁG (részletek: docs/KETLEPCSOS_HITELESITES.md) - Az átmeneti süti NEM munkamenet: a security.get_current_user() visszautasítja (typ mező). - A süti `SameSite=Strict`, `HttpOnly`, `Secure`, és csak a /mfa útvonalra érvényes; 5 percig él. - A kód egyszer használható (mfa_utils: atomi, visszajátszás elleni rögzítés). - A hibaüzenet mindig ugyanaz ("hibás vagy lejárt kód"), nem árulja el, melyik része volt rossz. - Az oldalak `Cache-Control: no-store` fejlécet kapnak. """ from fastapi import APIRouter, Request, Depends, Form, status from fastapi.responses import HTMLResponse, RedirectResponse from sqlalchemy.orm import Session from datetime import datetime, timezone import html import models import email_sor import login_vedelem import mentokod_utils import jelszo_hazirend import mfa_utils from config import ACCESS_TOKEN_EXPIRE_MINUTES from database import get_db from security import ( create_access_token, belepesi_token, decode_mfa_pending_token, get_client_ip, MFA_PENDING_COOKIE, ) from templates_env import templates router = APIRouter() _FEJLECEK = {"Cache-Control": "no-store"} # A beállító oldal a titkot (QR + kézi kulcs) mutatja: ne kerüljön gyorsítótárba, és a Referer ne vigye tovább _BEALLITAS_FEJLECEK = {"Cache-Control": "no-store", "Referrer-Policy": "no-referrer"} HIBAS_KOD_UZENET = "Hibás vagy lejárt kód. Ellenőrizd, hogy a telefonod ideje pontos-e, és a legfrissebb kódot írd be!" def _felhasznalo_az_atmeneti_tokenbol(request: Request, db: Session, cel: str = "kod"): """A félig bejelentkezett felhasználó, ha az átmeneti süti érvényes, a megfelelő CÉLÚ ("kod" / "beallitas": a két végpont nem használhatja egymás tokenjét), és a fiók még használható; különben None.""" payload = decode_mfa_pending_token(request.cookies.get(MFA_PENDING_COOKIE)) if not payload or payload.get("cel", "kod") != cel: return None user = db.query(models.User).filter(models.User.email == payload.get("sub")).first() if not user or not user.aktiv or user.aktivalva is False: return None # Ha a jelszó a token kiállítása óta megváltozott (vagy a fiókot visszaállították), a token már nem érvényes # (ugyanaz a szabály, mint a get_current_user()-ben). if user.jelszo_valtozott_at is not None: iat = payload.get("iat") if iat is None or datetime.fromtimestamp(iat, tz=timezone.utc).replace(tzinfo=None) < user.jelszo_valtozott_at: return None return user def _vissza_a_belepeshez() -> RedirectResponse: resp = RedirectResponse(url="/login", status_code=status.HTTP_303_SEE_OTHER) resp.delete_cookie(MFA_PENDING_COOKIE, path="/mfa") return resp def _oldal(request: Request, user, hiba: str = None): return templates.TemplateResponse( request=request, name="mfa.html", context={"email": user.email, "hiba": hiba}, headers=_FEJLECEK, ) @router.get("/mfa", response_class=HTMLResponse) def mfa_oldal(request: Request, db: Session = Depends(get_db)): user = _felhasznalo_az_atmeneti_tokenbol(request, db) if not user: return _vissza_a_belepeshez() return _oldal(request, user) @router.post("/mfa") def mfa_kod_ellenorzese( request: Request, kod: str = Form(""), db: Session = Depends(get_db) ): user = _felhasznalo_az_atmeneti_tokenbol(request, db) if not user: return _vissza_a_belepeshez() client_ip = get_client_ip(request) email = user.email # Ugyanaz a tiltás vonatkozik rá, mint a jelszó-találgatásra (azonos számlálók és határok). if login_vedelem.tiltva_e(db, client_ip, email): return _oldal(request, user, login_vedelem.TILTOTT_UZENET) # A szabály időközben megváltozhatott (pl. a rendszeradmin kikapcsolta, vagy a beállítást törölték): ilyenkor # nem erre az oldalra tartozik - újra a belépéshez, ott a friss szabály dönt. if mfa_utils.allapot(db, user) != mfa_utils.ALLAPOT_KESZ: return _vissza_a_belepeshez() # A próbálkozást az ELLENŐRZÉS ELŐTT számoljuk (külső átnézés találata): ha csak a hibás kód UTÁN rögzítenénk, # egy párhuzamos roham minden kérése átmenne a tiltás-ellenőrzésen, mielőtt bármelyik rögzítette volna a # sikertelenségét - így a tiltás határa jóval átléphető lenne (3 workeren még inkább). Siker esetén a # sikeres_torlese() úgyis törli ennek az e-mailnek az összes kísérletét, így a jó kód nem "romlik el" miatta. # Ha ez a kísérlet éri el a határt, a tiltás azonnal létrejön, és ez a kérés (akár jó a kód) is elutasításra kerül. user_id, company_id = user.id, user.company_id if login_vedelem.sikertelen_rogzitese(db, client_ip, email, user) is not None: return _oldal(request, user, login_vedelem.TILTOTT_UZENET) user = db.get(models.User, user_id) # a rögzítés rollback-et végzett: friss objektum if not mfa_utils.kod_ellenorzese_es_rogzitese(db, user, kod): return _oldal(request, user, HIBAS_KOD_UZENET) # Siker: az elfogadott időlépést AZONNAL commitoljuk (a sorzár ne tartson tovább, és a kód ne legyen újra felhasználható). db.commit() login_vedelem.sikeres_torlese(db, client_ip, email) db.add(models.AuditLog( company_id=company_id, user_id=user_id, field_name="Bejelentkezés", old_value="-", new_value=f"Sikeres bejelentkezés kétlépcsős hitelesítéssel ({client_ip})" )) db.commit() return _teljes_belepes(db, email, client_ip) def _teljes_belepes(db: Session, email: str, client_ip: str, resp=None): """A teljes munkamenet kiadása a 2FA-kód (vagy mentőkód) elfogadása után (belépés vagy első beállítás). Új munkamenet-azonosítót (sid) indít: ez dobja ki a fiók esetleges másik, még élő munkamenetét (v32, docs/EGYIDEJU_BELEPES.md); a rendszeradminra ez nem vonatkozik. `resp` (v34): ha meg van adva, a sütiket erre a (pl. a mentőkódokat mutató oldal) válaszra teszi, különben átirányítást ad a modulválasztóra.""" user = db.query(models.User).filter(models.User.email == email).first() if resp is None: # v36: ha a jelszó a következő napokban lejár, a belépés a Profilom jelszó-kártyájára visz (jelszo_hazirend.belepes_utani_cel) resp = RedirectResponse(url=jelszo_hazirend.belepes_utani_cel(db, user), status_code=status.HTTP_303_SEE_OTHER) resp.delete_cookie(MFA_PENDING_COOKIE, path="/mfa") resp.set_cookie( key="access_token", # "mfa": True - a get_current_user() kötelező 2FA mellett csak az ilyen jelzésű tokent fogadja el. value=belepesi_token(db, user, client_ip, mfa=True), httponly=True, secure=True, samesite="lax", max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60 ) return resp @router.post("/mfa/megse") def mfa_megse(): """Mégse: az átmeneti süti törlése, vissza a bejelentkezéshez (POST: állapotot módosít - a sütit törli).""" return _vissza_a_belepeshez() # ============================================================================================= # AZ ELSŐ BEÁLLÍTÁS (3. fázis) # ============================================================================================= def _kulcs_csoportositva(titok: str) -> str: """A kézi beírású kulcs olvashatóbb alakja (4 karakteres csoportok) - az alkalmazások a szóközt figyelmen kívül hagyják.""" return " ".join(titok[i:i + 4] for i in range(0, len(titok), 4)) def _beallitas_oldal(request: Request, user, titok: str, hiba: str = None): return templates.TemplateResponse( request=request, name="mfa_beallitas.html", context={ "email": user.email, "hiba": hiba, "qr": mfa_utils.qr_png_data_uri(mfa_utils.provisioning_uri(user, titok)), "kulcs": _kulcs_csoportositva(titok), }, headers=_BEALLITAS_FEJLECEK, ) def _beallitando_felhasznalo(request: Request, db: Session): """A beállításra jogosult (érvényes, "beallitas" célú átmeneti süti ÉS "beallitando" állapot) felhasználó, vagy None.""" user = _felhasznalo_az_atmeneti_tokenbol(request, db, cel="beallitas") if user is None or mfa_utils.allapot(db, user) != mfa_utils.ALLAPOT_BEALLITANDO: return None return user def _idegen_eredet(request: Request) -> bool: """Mélységi védelem a böngészős CSRF ellen - azonos a routers/rendszeradmin_mfa.py `_idegen_eredet`-ével (onnan körkörös import miatt nem importálható; ha az egyiket módosítod, a másikat is!). A `Sec-Fetch-Site` `same-origin` / `none` jó, minden más (`cross-site`, `same-site`) elutasítva; hiányzó fejléc (régi böngésző) nem utasítandó el. A fő védelem itt amúgy is az átmeneti süti: SameSite=Strict és Path=/mfa.""" sfs = request.headers.get("sec-fetch-site") return sfs is not None and sfs not in ("same-origin", "none") @router.get("/mfa/beallitas", response_class=HTMLResponse) def mfa_beallitas_oldal(request: Request, db: Session = Depends(get_db)): """A beállítás BEVEZETŐ oldala: magyarázat + "QR-kód létrehozása" gomb. SZÁNDÉKOSAN CSAK MEGJELENÍT, NEM ÍR (v28). Korábban (v27-ig) ez a GET minden megnyitáskor ÚJ titkot írt az adatbázisba (commit), így egy frissítés, a vissza-gomb vagy egy böngésző-előtöltés észrevétlenül érvénytelenítette a már beolvasott QR-kódot. Egy GET-nek biztonságosnak és ismételhetőnek kell lennie (docs/GET_ATTEKINTES.md, 7. fejezet), ezért az írás átkerült a POST /mfa/beallitas/inditas végpontra. A biztonsági tulajdonság VÁLTOZATLAN: minden MEGJELENÍTETT titok friss, a korábbi megerősítetlen eldobva (lásd lent).""" user = _beallitando_felhasznalo(request, db) if not user: return _vissza_a_belepeshez() return templates.TemplateResponse( request=request, name="mfa_beallitas.html", context={"email": user.email, "hiba": None, "qr": None, "kulcs": None}, # qr=None: a sablon a bevezetőt mutatja headers=_BEALLITAS_FEJLECEK, ) @router.post("/mfa/beallitas/inditas") def mfa_beallitas_inditasa(request: Request, db: Session = Depends(get_db)): """A beállítás INDÍTÁSA: új titkot készít és a QR-oldalt MEGJELENÍTI (a válasz közvetlenül ez az oldal, nincs átirányítás - a titok így sosem kerül URL-be vagy a böngészőelőzménybe; Cache-Control: no-store, Referrer-Policy: no-referrer). MINDEN indítás ÚJ titkot készít, a korábban meg nem erősítettet eldobja (biztonsági review-ból, 5c.2). Ha a meg nem erősített titkot újrahasználnánk, egy jelszót ismerő támadó (aki előbb indítja a beállítást) ugyanazt a titkot látná, mint a jogos felhasználó, és a felhasználó megerősítése után tartósan, észrevétlenül kódot tudna generálni. Így a támadó által látott titok érvénytelenedik, amint a felhasználó elindítja (és fordítva: a felhasználó hibás kódja jelzi, ha valaki közben újat készített). Ára: az oldal újratöltése (a böngésző az újraküldést megerősíttesse) új QR-t ad - a már beolvasott régi nem működik (az oldal erre figyelmeztet). Megerősített beállítás ide nem juthat (a _beallitando_felhasznalo kizárja). Az írás ATOMI (mfa_utils.titok_inditasa: UPDATE ... WHERE a beállítás nincs megerősítve): egy párhuzamos megerősítés mellett sem írhat felül megerősített beállítást.""" user = _beallitando_felhasznalo(request, db) if not user: return _vissza_a_belepeshez() if _idegen_eredet(request): return RedirectResponse(url="/mfa/beallitas", status_code=status.HTTP_303_SEE_OTHER) # nincs írás, a bevezetőre vissza user_id = user.id titok = mfa_utils.titok_inditasa(db, user_id) if titok is None: return _vissza_a_belepeshez() return _beallitas_oldal(request, db.get(models.User, user_id), titok) @router.post("/mfa/beallitas") def mfa_beallitas_megerositese( request: Request, kod: str = Form(""), db: Session = Depends(get_db) ): user = _beallitando_felhasznalo(request, db) if not user: return _vissza_a_belepeshez() client_ip = get_client_ip(request) email = user.email if login_vedelem.tiltva_e(db, client_ip, email): titok = mfa_utils.titok_olvasasa(user) return _beallitas_oldal(request, user, titok, login_vedelem.TILTOTT_UZENET) if titok else _vissza_a_belepeshez() titok = mfa_utils.titok_olvasasa(user) if titok is None: return RedirectResponse(url="/mfa/beallitas", status_code=status.HTTP_303_SEE_OTHER) # még nincs titok: a bevezető oldal (gomb) indítja # Ugyanaz a kód-találgatás elleni védelem, mint a /mfa-nál: a próbálkozást az ELLENŐRZÉS ELŐTT számoljuk. user_id, company_id = user.id, user.company_id if login_vedelem.sikertelen_rogzitese(db, client_ip, email, user) is not None: user = db.get(models.User, user_id) return _beallitas_oldal(request, user, mfa_utils.titok_olvasasa(user) or titok, login_vedelem.TILTOTT_UZENET) user = db.get(models.User, user_id) if not mfa_utils.megerosites(db, user, kod): db.rollback() user = db.get(models.User, user_id) return _beallitas_oldal(request, user, mfa_utils.titok_olvasasa(user) or titok, HIBAS_KOD_UZENET) # Siker: a beállítás KÉSZ (totp_megerositve) és az elfogadott időlépés rögzítve - egy tranzakcióban, azonnal commit. db.add(models.AuditLog( company_id=company_id, user_id=user_id, field_name="Kétlépcsős hitelesítés", old_value="nincs beállítva", new_value=f"Beállítva (hitelesítő alkalmazás, {client_ip})" )) db.commit() login_vedelem.sikeres_torlese(db, client_ip, email) _ertesito_level(db, user_id, client_ip) db.add(models.AuditLog( company_id=company_id, user_id=user_id, field_name="Bejelentkezés", old_value="-", new_value=f"Sikeres bejelentkezés (első 2FA-beállítás után) ({client_ip})" )) db.commit() return _belepes_mentokodokkal(request, db, user_id, company_id, email, client_ip) # ============================================================================================= # MENTŐKÓDOK (v34): az oldal, és az első beállítás végén a készlet átadása # ============================================================================================= def mentokod_oldal(request: Request, *, email: str, kodok=None, maradt: int = 0, hiba: str = None, ok: str = None, tovabb: str = "/valassz-modult", kezelheto: bool = True, elso: bool = False, kod_szoveg: str = ""): """A mentőkódok oldala (templates/mfa_mentokodok.html). `kodok`: az ÚJ készlet (egyszer látható) - ilyenkor ezt mutatja; különben a kezelő állapotot (hátralévő darabszám + "új készlet" űrlap). A válasz sosem kerülhet gyorsítótárba, a Referer nem viheti tovább.""" return templates.TemplateResponse( request=request, name="mfa_mentokodok.html", context={"email": email, "kodok": kodok, "maradt": maradt, "hiba": hiba, "ok": ok, "tovabb": tovabb, "kezelheto": kezelheto, "elso": elso, "darab": mentokod_utils.DARAB, "kuszob": mentokod_utils.FIGYELMEZTETO_KUSZOB, "kod_szoveg": kod_szoveg}, headers=_BEALLITAS_FEJLECEK, ) def mentokod_szoveg(email: str, kodok) -> str: """A letölthető .txt tartalma.""" return ("EmeLOG mentőkódok - " + email + "\n" "Minden kód csak egyszer használható. Tartsd biztonságos helyen (pl. jelszókezelő, kinyomtatva) - ne a telefonodon, a hitelesítő alkalmazás mellett!\n\n" + "\n".join(kodok) + "\n") def _belepes_mentokodokkal(request: Request, db: Session, user_id: int, company_id, email: str, client_ip: str): """Az ELSŐ 2FA-beállítás végén: új mentőkód-készlet, és a teljes munkamenet kiadása egy olyan oldalon, amely a kódokat EGYSZER megmutatja (a "Tovább" gomb a modulválasztóra visz). Ha a készlet elkészítése bármiért elbukik, a belépést nem akadályozza: a felhasználó később is kérhet készletet (Beállítások -> Mentőkódok).""" try: kodok = mentokod_utils.uj_keszlet(db, user_id) db.add(models.AuditLog( company_id=company_id, user_id=user_id, field_name="Kétlépcsős hitelesítés - mentőkódok", old_value="-", new_value=f"Új mentőkód-készlet ({mentokod_utils.DARAB} db, első beállítás, {client_ip})" )) db.commit() except Exception as ex: # pragma: no cover - védőháló db.rollback() print(f"⚠️ A mentőkód-készlet elkészítése sikertelen: {type(ex).__name__}: {ex}") return _teljes_belepes(db, email, client_ip) oldal = mentokod_oldal(request, email=email, kodok=kodok, elso=True, kod_szoveg=mentokod_szoveg(email, kodok)) return _teljes_belepes(db, email, client_ip, oldal) def _ertesito_level(db: Session, user_id: int, client_ip: str) -> None: """Biztonsági értesítő a felhasználónak: a kétlépcsős hitelesítés beállítva lett. Ha NEM ő volt (pl. egy ellopott jelszóval valaki más állította be az első faktort), ebből észreveszi. Nem akadályozhatja a belépést: hibánál csak a konzolra ír. A kategória "rendszer" (a levél-feliratkozásokkal nem kapcsolható ki).""" try: user = db.get(models.User, user_id) if user is None or not user.email: return e = lambda x: html.escape("" if x is None else str(x), quote=True) # noqa: E731 most = login_vedelem.helyi_ido_szoveg(login_vedelem._utc_most()) torzs = ( "
" "

Kétlépcsős hitelesítés beállítva

" f"

Kedves {e(user.username)}!

" "

Az EmeLOG fiókodon beállították a kétlépcsős hitelesítést (hitelesítő alkalmazás).

" f"

Időpont: {e(most)}
IP-cím: {e(client_ip)}

" "

Ha nem te voltál, azonnal értesítsd a rendszergazdát - valaki ismerheti a jelszavadat!

" ) email_sor.sorba_allit([user.email], "Kétlépcsős hitelesítés beállítva az EmeLOG fiókodon", torzs, kategoria="rendszer", prioritas=email_sor.PRIORITAS_NORMAL, company_id=user.company_id, db=db) db.commit() except Exception as ex: db.rollback() print(f"⚠️ A 2FA-beállítás értesítő levelét nem sikerült sorba állítani: {type(ex).__name__}: {ex}")