import os import shutil from datetime import datetime, timezone from fastapi import APIRouter, Request, Depends, Form, status, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse, FileResponse from sqlalchemy.orm import Session, joinedload import models import aktivalas_utils import mfa_utils import jelszo_hazirend from database import get_db from security import ( sanitize_filename, verify_password, get_password_hash, get_current_user, get_working_company_id, is_company_active, is_rendszeradminisztrator, is_safe_extension, jelszo_eleg_eros_e, tartalom_illeszkedik_e_kiterjesztesre, is_company_allowed ) from licenc_utils import licenc_tullepne_uj_felhasznalonal, licenc_reaktivalas_hiba from naplo_utils import naploz from templates_env import templates from routers.rendszeradmin_felhasznalok import SZEREPKOROK router = APIRouter() # KÖZEPES (2026-10-03-i külső biztonsági audit, 2. kör): a /uj-ceges-felhasznalo végpont eddig # a "role" és "hozzaferes_szint" mezőket validálatlanul fogadta a nyers form-adatból - egy # cég-Admin elméletben a sablonban felkínált 3 szerepkörön (Felhasználó/EÜI/VIVU) kívül BÁRMIT # beírhatott volna (pl. "Admin", "Emelőgép szakértő"), a hozzaferes_szint pedig ismeretlen érték # esetén csendben a legtágabb ("CEG", teljes cég) hozzáférést eredményezte volna. ENGEDELYEZETT_UJ_FELHASZNALO_SZEREPKOROK = ["Felhasználó", "Emelőgép ügyintéző", "VIVU"] ENGEDELYEZETT_HOZZAFERES_SZINTEK = ["CEG", "TELEPHELY", "GEP"] # J-18 (v66): a Tűzvédelem oldal "Új karbantartó" űrlapja role=Karbantartó-t küld, amit a fenti fehérlista eddig elutasított # (a szerepkör a felületről nem volt létrehozható). Külön állandó, mert NEM minden cégnél értelmes: csak a TUZVEDELEM modulra # előfizetett cégnél engedett, és a Karbantartó CSAK a Tűzvédelem modult kapja (mint a VIVÜ a sajátját). TUZVEDELMI_KARBANTARTO_SZEREPKOR = "Karbantartó" # J-13 (v69): a céges Admin által megadott telephely-/gép-korlátozás ELLENŐRZÉSE. Eddig a /szerkeszt-felhasznalo és az # /uj-ceges-felhasznalo (1) a hozzaferes_szint-et (szerkesztésnél) fehérlista nélkül fogadta - egy ismeretlen érték az # is_machine_allowed végén `return True`-ra, azaz teljes céges elérésre esett -, és (2) az allowed_sites/allowed_machines # azonosítókat CÉG-SZŰRÉS NÉLKÜL kereste meg, így idegen cég telephelye/gépe is a felhasználóhoz köthető volt. # A minta ugyanaz, mint az RA-oldali teljes_szerkesztés-nél (rendszeradmin_felhasznalok.py): fehérlista + a felhasználó cégére # szűrt keresés + darabszám-egyezés; bármi eltérés -> 400, és SEMMI sem mentődik. def hozzaferes_alanyok_ellenorzese(db: Session, ceg_id: int, szint, nyers_telephelyek, nyers_gepek): """(telephelyek, gepek, hibauzenet). Hibánál az első két érték None, a harmadik a 400-as válasz üzenete. CEG szinten a lista üres (a korlátozás nélküli hozzáférés); TELEPHELY-nél csak a telephelyek, GEP-nél csak a gépek számítanak.""" if szint not in ENGEDELYEZETT_HOZZAFERES_SZINTEK: return None, None, "Érvénytelen hozzáférési szint!" try: th_idk = [int(x) for x in nyers_telephelyek] gep_idk = [int(x) for x in nyers_gepek] except (TypeError, ValueError): return None, None, "Érvénytelen telephely/gép-azonosító!" telephelyek, gepek = [], [] if szint == "TELEPHELY" and th_idk: telephelyek = db.query(models.Site).filter(models.Site.id.in_(th_idk), models.Site.company_id == ceg_id).all() if len(telephelyek) != len(set(th_idk)): return None, None, "A telephelyek között van olyan, ami nem ehhez a céghez tartozik!" elif szint == "GEP" and gep_idk: gepek = db.query(models.Machine).filter(models.Machine.id.in_(gep_idk), models.Machine.company_id == ceg_id).all() if len(gepek) != len(set(gep_idk)): return None, None, "A gépek között van olyan, ami nem ehhez a céghez tartozik!" return telephelyek, gepek, None # J-19 (v66): az /uj-admin-felhasznalo (csak rendszeradmin) szerepkör-fehérlistája - ugyanaz a három, amit a felhasznalok.html felkínál. RA_ENGEDELYEZETT_UJ_FELHASZNALO_SZEREPKOROK = ["Admin", "Emelőgép szakértő", "Emelőgép vizsgáló"] # A /felhasznalok oldalon a SZERKESZTÉS/JELSZÓ/INAKTIVÁLÁS gombok csak azoknál a soroknál jelennek meg, # amelyekre a 2026-10-04-i (v24) bővítés ELŐTT is megjelentek: cég-Admin és külsős Emelőgép szakértő/vizsgáló # fiókok, a rendszercégen (company_id == 1) KÍVÜL. A többi sor (Felhasználó, EÜI, VIVU, EBK, Megtekintő, # Karbantartó, a rendszeradminisztrátorok és az ismeretlen/régi szerepkörök) ebben a szeletben CSAK OLVASHATÓ. KEZELHETO_SZEREPKOROK = ("Admin", "Emelőgép szakértő", "Emelőgép vizsgáló") def osszes_felhasznalo_sorok(db: Session) -> list: """A rendszeradminisztrátor teljes felhasználólistájának sorai (/felhasznalok, v24, 1. szelet). FIGYELEM - JOGOSULTSÁG: ezt a függvényt KIZÁRÓLAG a /felhasznalok végpont hívhatja, a `is_rendszeradminisztrator()` ellenőrzése UTÁN. Ő a rendszerben az egyetlen szerepkör, aki a cégektől függetlenül MINDEN felhasználót láthat. A cég-Admin továbbra sincs ezen az oldalon (átirányítás a /lista oldalra), az ő saját cége felhasználóit a /beallitasok oldal mutatja, ez NEM változott. Soronként egy szótárat ad vissza (a sablon ezt jeleníti meg): u - a User objektum (a sablon a hashed_password / totp_* oszlopokat SOSEM írja ki!) ra - rendszeradminisztrátor-e (Admin + company_id == 1) kezelheto - megjelennek-e a szerkesztés/jelszó/inaktiválás gombok (lásd KEZELHETO_SZEREPKOROK) modulok - a felhasználó TÉNYLEGESEN elérhető modulkódjai (cég ∩ saját; RA-nak minden aktív modul), ugyanaz a szabály, mint a security.get_user_modul_kodok()-ban modulok_sajat - a felhasználóhoz közvetlenül rendelt modulkódok (a cég-metszet nélkül) mfa - a 2FA állapota a cég szabálya szerint (ki / beallitando / kesz / helyreallitando) mfa_beallitva - "kesz" / "helyreallitando" / "nincs": a felhasználó SAJÁT beállítási állapota telephely - a telephely_hozzaferes_id-hez tartozó telephely neve (ha van) utolso_ceg - a legutóbb kiválasztott munkakörnyezet cégének neve (ha van) A titkos adatokat (jelszó-hash, TOTP titok) a sor SZÁNDÉKOSAN csak a mfa_utils állapot-függvényein keresztül érinti, a tényleges értékük sosem kerül a sablonba. """ aktiv_modulok = {m.kod for m in db.query(models.Modul).filter(models.Modul.aktiv == True).all()} # noqa: E712 cegek = db.query(models.Company).options(joinedload(models.Company.engedelyezett_modulok)).all() ceg_modulok = {c.id: {m.kod for m in c.engedelyezett_modulok if m.aktiv} for c in cegek} ceg_nevek = {c.id: c.nev for c in cegek} telephely_nevek = {s.id: s.nev for s in db.query(models.Site.id, models.Site.nev).all()} felhasznalok = ( db.query(models.User) .options( joinedload(models.User.ceg), joinedload(models.User.allowed_companies), joinedload(models.User.allowed_sites), joinedload(models.User.allowed_machines), joinedload(models.User.sajat_modulok), ) .order_by(models.User.company_id, models.User.role, models.User.username) .all() ) sorok = [] for u in felhasznalok: ra = is_rendszeradminisztrator(u) sajat = {m.kod for m in u.sajat_modulok if m.aktiv} modulok = aktiv_modulok if ra else (ceg_modulok.get(u.company_id, set()) & sajat) sorok.append({ "u": u, "ra": ra, "kezelheto": u.role in KEZELHETO_SZEREPKOROK and u.company_id != 1, "modulok": sorted(modulok), "modulok_sajat": sorted(sajat), "mfa": mfa_utils.allapot(db, u), "mfa_beallitva": mfa_utils.beallitasi_allapot(u), "telephely": telephely_nevek.get(u.telephely_hozzaferes_id), "utolso_ceg": ceg_nevek.get(u.utolso_ceg_id), }) return sorok @router.get("/felhasznalok", response_class=HTMLResponse) def felhasznalok_oldal( request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): # JOGOSULTSÁG (változatlan): csak a rendszeradminisztrátor. Mindenki más a /lista oldalra kerül. if not is_rendszeradminisztrator(current_user): return RedirectResponse(url="/lista") # v24: a lista már a rendszer MINDEN felhasználóját mutatja (korábban csak a cég-Admin és a külsős # szakértő/vizsgáló fiókokat, a rendszercég nélkül) - lásd osszes_felhasznalo_sorok(). sorok = osszes_felhasznalo_sorok(db) cegek = db.query(models.Company).filter(models.Company.id != 1).all() return templates.TemplateResponse( request=request, name="felhasznalok.html", context={ "sorok": sorok, "users": [s["u"] for s in sorok], "user": current_user, "cegek": cegek, "mind_cegek": db.query(models.Company).order_by(models.Company.nev).all(), # v24, 2. szelet: a szerepkör/cég módosító modál választékai (csak AKTÍV cég, a rendszercég nélkül; a 9 ismert szerepkör) "valaszthato_cegek": db.query(models.Company).filter(models.Company.id != 1, models.Company.aktiv == True).order_by(models.Company.nev).all(), # noqa: E712 "valaszthato_szerepkorok": SZEREPKOROK, "szerepkorok": sorted({s["u"].role for s in sorok if s["u"].role}), } ) @router.get("/uploads/{company_id}/bizonyitvanyok/{filename}") def bizonyitvany_fajl_kiszolgalas( company_id: int, filename: str, current_user: models.User = Depends(get_current_user) ): """A bizonyítvány feltöltése/cseréje eddig is működött, de erre az útvonal-mintára nem volt kiszolgáló route - a szerkesztő ablakban megjelenő "Jelenlegi fájl megtekintése" link mindig 404-et adott volna (ugyanaz a hibaosztály, mint az arculati logónál, csak itt még nem jelentette a user - 2026-09-30-i logó-hiba nyomán proaktívan átnézve és javítva).""" if current_user.role != "Admin" or not is_company_allowed(current_user, company_id): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Nincs jogosultságod a kért fájl eléréséhez!") safe_filename = sanitize_filename(filename) file_path = f"uploads/{company_id}/bizonyitvanyok/{safe_filename}" if not os.path.exists(file_path): raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="A keresett fájl nem található a szerveren.") return FileResponse(file_path) @router.post("/uj-admin-felhasznalo") async def uj_admin_felhasznalo( request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if not is_rendszeradminisztrator(current_user): return JSONResponse(status_code=403, content={"message": "Nincs jogosultságod!"}) if current_user.csak_olvasas: return JSONResponse(status_code=403, content={"message": "Csak megtekintési jogosultsággal rendelkezel, szerkesztés nem lehetséges!"}) form_data = await request.form() username = form_data.get("username") email = form_data.get("email") role = form_data.get("role") try: target_company_id = int(form_data.get("target_company_id")) except (TypeError, ValueError): return JSONResponse(status_code=400, content={"message": "Érvénytelen cég-azonosító!"}) # J-19 (v66): a szerepkör eddig validálatlan volt (a testvér /uj-ceges-felhasznalo fehérlistás), és az 1-es (rendszer-) cégbe is # lehetett volna "Admin"-t felvenni - azaz rendszeradminisztrátort, a rendszeradmin-kezelés (utolsó aktív RA védelem stb.) megkerülésével. if role not in RA_ENGEDELYEZETT_UJ_FELHASZNALO_SZEREPKOROK: return JSONResponse(status_code=400, content={"message": "Érvénytelen szerepkör!"}) if target_company_id == 1: return JSONResponse(status_code=400, content={"message": "A rendszer-cégbe (rendszeradminisztrátorok) ezen az úton nem vehető fel felhasználó!"}) if not is_company_active(target_company_id, db): return JSONResponse(status_code=403, content={"message": "Ez a cég inaktív státuszú, új felhasználó nem vehető fel hozzá!"}) if not aktivalas_utils.email_ervenyes_e(email): users = db.query(models.User).filter( models.User.role.in_(["Admin", "Emelőgép szakértő", "Emelőgép vizsgáló"]), models.User.company_id != 1 ).all() cegek = db.query(models.Company).filter(models.Company.id != 1).all() return templates.TemplateResponse( request=request, name="felhasznalok.html", context={"users": users, "user": current_user, "cegek": cegek, "error": "Érvénytelen e-mail cím! Az aktiváló levelet erre a címre küldjük."} ) letezo = db.query(models.User).filter(models.User.email == email).first() if letezo: users = db.query(models.User).filter( models.User.role.in_(["Admin", "Emelőgép szakértő", "Emelőgép vizsgáló"]), models.User.company_id != 1 ).all() cegek = db.query(models.Company).filter(models.Company.id != 1).all() return templates.TemplateResponse( request=request, name="felhasznalok.html", context={"users": users, "user": current_user, "cegek": cegek, "error": "Ez az e-mail cím már foglalt!"} ) file_path = None bizonyitvany_file = form_data.get("bizonyitvany_file") if bizonyitvany_file: if getattr(bizonyitvany_file, "filename", "") != "": if not is_safe_extension(bizonyitvany_file.filename): return JSONResponse(status_code=400, content={"message": "Nem engedélyezett fájlformátum a bizonyítványnál!"}) _fejresz = bizonyitvany_file.file.read(8) bizonyitvany_file.file.seek(0) if not tartalom_illeszkedik_e_kiterjesztesre(bizonyitvany_file.filename, _fejresz): return JSONResponse(status_code=400, content={"message": "A feltöltött fájl tartalma nem egyezik a kiterjesztésével!"}) mappa = f"uploads/{target_company_id}/bizonyitvanyok" os.makedirs(mappa, exist_ok=True) # Az e-mail cím a fájlnév részét képezi, ezért ezt is a fájlnévvel egyező módon # tisztítani kell - máskülönben egy "../../valami" tartalmú e-mail cím (a mezőt # semmi nem korlátozza szigorúan e-mail formátumra a beküldés pillanatában) # könyvtár-áthágást (path traversal) tenne lehetővé. safe_email = sanitize_filename(email) safe_filename = sanitize_filename(bizonyitvany_file.filename) file_path = f"/{mappa}/{safe_email}_{safe_filename}" with open(file_path.lstrip('/'), "wb+") as f: shutil.copyfileobj(bizonyitvany_file.file, f) uj_user = models.User( username=username, email=email, # AKTIVÁLÁS (E szelet): nincs kezdőjelszó - a felhasználó az aktiváló linken állítja be (aktivalas_utils.py) hashed_password=aktivalas_utils.jelszo_nelkuli_hash(), role=role, company_id=target_company_id, hozzaferes_szint="CEG", bizonyitvany_szam=form_data.get("bizonyitvany_szam", ""), kamarai_azonosito=form_data.get("kamarai_azonosito", ""), engedely_szam=form_data.get("engedely_szam", ""), tanusitvany_szam=form_data.get("tanusitvany_szam", ""), bizonyitvany_path=file_path, csak_olvasas=(form_data.get("csak_olvasas") == "on") ) if role in ("Emelőgép szakértő", "Emelőgép vizsgáló"): ceg_idek = form_data.getlist("allowed_companies") for cid in ceg_idek: comp_obj = db.query(models.Company).filter(models.Company.id == int(cid)).first() if comp_obj: uj_user.allowed_companies.append(comp_obj) # A cég ÖSSZES előfizetett modulját megkapja - enélkül az új (céges Admin vagy külsős # szakértő/vizsgáló) felhasználó a bejelentkezés után rögtön elakadna a modulválasztó # oldalon (0 modul), egészen a következő szerver-újraindításig. celzott_ceg = db.query(models.Company).filter(models.Company.id == target_company_id).first() uj_user.sajat_modulok = list(celzott_ceg.engedelyezett_modulok) if celzott_ceg else [] db.add(uj_user) aktivalas_utils.meghivas(db, uj_user, current_user) # aktiváló levél (E szelet) # ÚJ (2026-10-02-i user-jelzés): a felhasználó-létrehozás eddig egyáltalán nem került be a # Jogosultsági naplóba (routers/settings.py beallitasok_oldal, templates/beallitasok.html # "Jogosultsági napló" tab) - csak az aktiválás/inaktiválás (l. kapcsol_felhasznalo_aktiv). db.add(models.AuditLog( company_id=target_company_id, user_id=current_user.id, field_name="Új felhasználó", old_value="-", new_value=f"{username} ({email}) - szerepkör: {role}" )) db.commit() return RedirectResponse(url="/felhasznalok", status_code=status.HTTP_303_SEE_OTHER) def _felhasznalo_pillanatkep(u) -> dict: """A szerkeszthető felhasználói adatok egyszerű szöveges pillanatképe (J-20, v77): a szerkesztés előtti és utáni állapot különbsége kerül mezőnként a naplóba ("Felhasználó adata: "). Jelszó / hash / token SOHA nincs benne.""" def igen_nem(v): return "igen" if v else "nem" return { "név": u.username, "e-mail": u.email, "csak olvasás": igen_nem(u.csak_olvasas), "dátum-szerkesztési jog": igen_nem(u.jog_datum_szerkesztes), "üzemóra-rögzítési jog": igen_nem(u.jog_uzemora_rogzites), "dokumentum-feltöltési jog": igen_nem(u.jog_dokumentum_feltoltes), "gép-szerkesztési jog": igen_nem(u.jog_gep_szerkesztes), "bizonyítvány száma": u.bizonyitvany_szam or "-", "bizonyítvány-fájl": u.bizonyitvany_path or "-", "kamarai azonosító": u.kamarai_azonosito or "-", "engedély száma": u.engedely_szam or "-", "tanúsítvány száma": u.tanusitvany_szam or "-", "orvosi alkalmasság érvényessége": str(u.orvosi_alkalmassag_ervenyesseg or "-"), "jogosítvány érvényessége": str(u.jogositvany_ervenyesseg or "-"), "hozzáférési szint": u.hozzaferes_szint or "-", "hozzáférés - telephelyek": ", ".join(sorted(s.nev for s in u.allowed_sites)) or "-", "hozzáférés - gépek": ", ".join(sorted(str(g.id) for g in u.allowed_machines)) or "-", } @router.post("/szerkeszt-felhasznalo") async def szerkeszt_felhasznalo( request: Request, user_id: int = Form(...), username: str = Form(...), email: str = Form(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): target_user = db.query(models.User).filter(models.User.id == user_id).first() if not target_user: return JSONResponse(status_code=404, content={"message": "Nem található!"}) # KRITIKUS (2026-09-28-i külső biztonsági audit): ez a végpont korábban csak azt ellenőrizte, # hogy a célfelhasználó ugyanahhoz a céghez tartozik-e - szerepkör-ellenőrzés NEM volt benne, # holott a szomszédos kapcsol_felhasznalo_aktiv/admin_jelszo_csere végpontok igen. Emiatt # BÁRMELYIK bejelentkezett, alacsony jogú "Felhasználó" a SAJÁT user_id-jével meghívva # magának adhatott jog_uzemora_rogzites/jog_dokumentum_feltoltes/jog_gep_szerkesztes jogokat, # és a hozzaferes_szint-jét "CEG"-re állíthatta (kitörölve a telephely-/gép-korlátozást) - ez # cégen belüli vertikális jogosultság-eszkaláció volt, önmagában, további feltétel nélkül # kihasználható. A minta pontosan a kapcsol_felhasznalo_aktiv-ét követi. if not is_rendszeradminisztrator(current_user) and current_user.role != "Admin": return JSONResponse(status_code=403, content={"message": "Nincs jogosultságod felhasználó szerkesztéséhez!"}) working_company_id = get_working_company_id(current_user, request) if not is_rendszeradminisztrator(current_user) and target_user.company_id != working_company_id: return JSONResponse(status_code=403, content={"message": "Nincs jogod ehhez a felhasználóhoz!"}) if not is_company_active(target_user.company_id, db): return JSONResponse(status_code=403, content={"message": "Ez a cég inaktív státuszú, a felhasználó adatai nem szerkeszthetők!"}) if current_user.csak_olvasas: return JSONResponse(status_code=403, content={"message": "Csak megtekintési jogosultsággal rendelkezel, szerkesztés nem lehetséges!"}) form_data = await request.form() # J-13 (v69): a hozzáférési szint + telephely-/gép-lista ellenőrzése MINDEN mellékhatás (fájl-írás, mezők) ELŐTT: # hibánál 400 és semmi sem mentődik. Az Admin szerepkörnél a blokk nem fut (neki nincs ilyen korlátozás). uj_telephelyek, uj_gepek = [], [] if target_user.role != "Admin" and "hozzaferes_szint" in form_data: uj_telephelyek, uj_gepek, hozzaferes_hiba = hozzaferes_alanyok_ellenorzese( db, target_user.company_id, form_data.get("hozzaferes_szint"), form_data.getlist("allowed_sites"), form_data.getlist("allowed_machines")) if hozzaferes_hiba: return JSONResponse(status_code=400, content={"message": hozzaferes_hiba}) elotte = _felhasznalo_pillanatkep(target_user) # J-20 (v77): a változás-napló alapja (a módosítások ELŐTT) if target_user.email != email: foglalt = db.query(models.User).filter(models.User.email == email).first() if foglalt: return JSONResponse(status_code=400, content={"message": "Az email cím már foglalt!"}) if not aktivalas_utils.aktivalva_e(target_user) and not aktivalas_utils.email_ervenyes_e(email): # váró fióknál az új címre új link megy - érvénytelen cím esetén a fiók örökre elakadna return JSONResponse(status_code=400, content={"message": "Az email cím formátuma érvénytelen!"}) regi_email = target_user.email target_user.email = email # Ha a fiók még aktiválásra vár: a régi címre küldött link érvényét veszti, az új címre új link megy aktivalas_utils.email_valtozas_kezelese(db, target_user, current_user, regi_email=regi_email) target_user.username = username if target_user.role != "Admin": target_user.csak_olvasas = (form_data.get("csak_olvasas") == "on") # Meglévő gépek dátumainak kézi szerkesztési joga - kizárólag ennél a 3 szerepkörnél # értelmezett, ezért itt, a többi jog_* mezőtől függetlenül kezeljük (az Admin szerepkör # ki van zárva a lenti nagyobb blokkból, itt viszont pont ő is az egyik célszerepkör). if target_user.role in ["Admin", "Emelőgép szakértő", "Emelőgép vizsgáló", "Emelőgép ügyintéző"]: target_user.jog_datum_szerkesztes = (form_data.get("jog_datum_szerkesztes") == "on") if target_user.role != "Admin": if "bizonyitvany_szam" in form_data: target_user.bizonyitvany_szam = form_data.get("bizonyitvany_szam") # ÚJ (2026-09-29-i user-jelzés): eddig a szerkesztő űrlapon NEM lehetett új bizonyítvány- # fájlt feltölteni/cserélni egy meglévő felhasználónál - csak a létrehozáskor (l. # uj_ceges_felhasznalo/uj_admin_felhasznalo). Ugyanazt az ellenőrzési mintát követi. bizonyitvany_file = form_data.get("bizonyitvany_file") if bizonyitvany_file and getattr(bizonyitvany_file, "filename", "") != "": if not is_safe_extension(bizonyitvany_file.filename): return JSONResponse(status_code=400, content={"message": "Nem engedélyezett fájlformátum a bizonyítványnál!"}) fejresz = bizonyitvany_file.file.read(8) bizonyitvany_file.file.seek(0) if not tartalom_illeszkedik_e_kiterjesztesre(bizonyitvany_file.filename, fejresz): return JSONResponse(status_code=400, content={"message": "A feltöltött fájl tartalma nem egyezik a kiterjesztésével!"}) mappa = f"uploads/{target_user.company_id}/bizonyitvanyok" os.makedirs(mappa, exist_ok=True) safe_email = sanitize_filename(target_user.email) safe_filename = sanitize_filename(bizonyitvany_file.filename) uj_path = f"/{mappa}/{safe_email}_{safe_filename}" with open(uj_path.lstrip('/'), "wb+") as f: shutil.copyfileobj(bizonyitvany_file.file, f) target_user.bizonyitvany_path = uj_path if "kamarai_azonosito" in form_data: target_user.kamarai_azonosito = form_data.get("kamarai_azonosito") if "engedely_szam" in form_data: target_user.engedely_szam = form_data.get("engedely_szam") if "tanusitvany_szam" in form_data: target_user.tanusitvany_szam = form_data.get("tanusitvany_szam") if form_data.get("jog_uzemora_rogzites") == "on": target_user.jog_uzemora_rogzites = True else: target_user.jog_uzemora_rogzites = False if form_data.get("jog_dokumentum_feltoltes") == "on": target_user.jog_dokumentum_feltoltes = True else: target_user.jog_dokumentum_feltoltes = False if form_data.get("jog_gep_szerkesztes") == "on": target_user.jog_gep_szerkesztes = True else: target_user.jog_gep_szerkesztes = False # --- ÚJ: HR DÁTUMOK FRISSÍTÉSE --- try: if form_data.get("orvosi_alkalmassag_ervenyesseg"): target_user.orvosi_alkalmassag_ervenyesseg = datetime.strptime(form_data.get("orvosi_alkalmassag_ervenyesseg"), "%Y-%m-%d").date() else: target_user.orvosi_alkalmassag_ervenyesseg = None if form_data.get("jogositvany_ervenyesseg"): target_user.jogositvany_ervenyesseg = datetime.strptime(form_data.get("jogositvany_ervenyesseg"), "%Y-%m-%d").date() else: target_user.jogositvany_ervenyesseg = None except ValueError: return JSONResponse(status_code=400, content={"message": "Érvénytelen dátumformátum!"}) # --------------------------------- if "hozzaferes_szint" in form_data: # J-13 (v69): a már ellenőrzött (a felhasználó cégére szűrt) lista kerül mentésre target_user.hozzaferes_szint = form_data.get("hozzaferes_szint") target_user.allowed_sites = uj_telephelyek target_user.allowed_machines = uj_gepek # J-20 (v77): minden VÁLTOZOTT mező külön naplósor ("Felhasználó adata: "), a módosítással egy tranzakcióban utana = _felhasznalo_pillanatkep(target_user) for mezo, regi in elotte.items(): naploz(db, current_user, target_user.company_id, f"Felhasználó adata: {mezo}", f"{elotte['név']}: {regi}", f"{elotte['név']}: {utana[mezo]}", csak_ha_valtozott=True) db.commit() referer_url = request.headers.get("referer") if not referer_url: referer_url = "/lista" return RedirectResponse(url=referer_url, status_code=status.HTTP_303_SEE_OTHER) @router.post("/kapcsol-felhasznalo-aktiv/{target_id}") def kapcsol_felhasznalo_aktiv( request: Request, target_id: int, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): """Felhasználó inaktiválása/reaktiválása. Törlés helyett - a történeti adatok (vizsgálatok, naplók) megmaradnak.""" target_user = db.query(models.User).filter(models.User.id == target_id).first() if not target_user or current_user.id == target_id: return JSONResponse(status_code=403, content={"message": "Hiba történt!"}) is_admin = is_rendszeradminisztrator(current_user) is_sajat_ceg_admin = (current_user.role == "Admin" and target_user.company_id == current_user.company_id) if current_user.csak_olvasas: return JSONResponse(status_code=403, content={"message": "Csak megtekintési jogosultsággal rendelkezel, szerkesztés nem lehetséges!"}) if is_admin or is_sajat_ceg_admin: if target_user.role == "VIVU" and not target_user.aktiv: # Reaktiválás előtt is ellenőrizni kell a "cégenként egy VIVÜ" szabályt, különben # egy régi, inaktivált VIVU-user visszakapcsolása megkerülné az uj-ceges-felhasznalo # létrehozáskori védelmét, és két aktív VIVU keletkezne egy cégnél. van_mar_vivu = db.query(models.User).filter( models.User.company_id == target_user.company_id, models.User.role == "VIVU", models.User.aktiv == True, models.User.id != target_user.id, ).first() if van_mar_vivu: return JSONResponse(status_code=400, content={ "message": f'Ennél a cégnél már van aktív VIVÜ ügyintéző ("{van_mar_vivu.username}") - ' f'cégenként csak egy lehet, ez a felhasználó nem aktiválható vissza.' }) # J-06 (v66): a licenc-limit a visszakapcsoláskor is él (eddig csak a létrehozásnál futott: 5/5 -> egyiket inaktiválja -> # újat felvesz -> a régit visszakapcsolja -> 6 aktív). licenc_hiba = licenc_reaktivalas_hiba(target_user, db) if licenc_hiba: return JSONResponse(status_code=400, content={"message": licenc_hiba}) regi_allapot = "Aktív" if target_user.aktiv else "Inaktív" target_user.aktiv = not target_user.aktiv uj_allapot = "Aktív" if target_user.aktiv else "Inaktív" naplo = models.AuditLog( company_id=target_user.company_id, user_id=current_user.id, field_name="Felhasználó állapota", old_value=f"{target_user.username}: {regi_allapot}", new_value=f"{target_user.username}: {uj_allapot}" ) db.add(naplo) if not target_user.aktiv: # Inaktivált (még aktiválásra váró) fiók aktiváló linkje ne maradjon élő - visszakapcsolás után új link kell aktivalas_utils.tokenek_torlese(db, target_user) db.commit() allapot = "aktiválva" if target_user.aktiv else "inaktiválva" return JSONResponse(status_code=200, content={"message": f"Felhasználó sikeresen {allapot}!", "aktiv": target_user.aktiv}) return JSONResponse(status_code=403, content={"message": "Hiba történt!"}) @router.post("/admin-jelszo-csere") async def admin_jelszo_csere( request: Request, target_id: int = Form(...), uj_jelszo: str = Form(...), uj_jelszo2: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): # v27: a kényszerített jelszócserénél is meg kell erősíteni az új jelszót (lásd sajat_jelszo_csere()). Az űrlapok # data-ajax-form-osak: a 400-as JSON "message" figyelmeztetésként megjelenik, siker esetén az oldal újratölt. if uj_jelszo != uj_jelszo2: return JSONResponse(status_code=400, content={"message": "A két új jelszó nem egyezik!"}) # J-07 (v66): ez a végpont MÁSOK jelszavát állítja, a régi jelszó ismerete nélkül. Önmagára nem működhet: ellopott admin-munkamenettel # különben a jogos felhasználó kizárható lenne. A saját jelszó cseréje a /sajat-jelszo-csere (Profilom), a régi jelszó megadásával. if target_id == current_user.id: return JSONResponse(status_code=400, content={ "message": "A saját jelszavadat nem itt, hanem a Profilom oldalon, a régi jelszó megadásával módosíthatod." }) target_user = db.query(models.User).filter(models.User.id == target_id).first() if target_user: is_admin = is_rendszeradminisztrator(current_user) is_sajat_ceg_admin = (current_user.role == "Admin" and target_user.company_id == current_user.company_id) if (is_admin or is_sajat_ceg_admin) and is_company_active(target_user.company_id, db) and not current_user.csak_olvasas \ and not aktivalas_utils.aktivalva_e(target_user): # AKTIVÁLÁSRA VÁRÓ fiók (E szelet): a jelszót a felhasználó állítja be az e-mailben kapott linken. # - a cég Admin-ja a "Jelszó módosítása" gombbal új aktiváló linket kérhet (a megadott jelszót nem használjuk); # - a rendszeradmin kézzel aktiválhatja a fiókot a megadott jelszóval (ha az e-mail nem érkezik meg). # (Így minden szerepkörnél - EBK/VIVÜ/Megtekintő is - van kiút, ha a levél elveszett.) if is_admin and jelszo_eleg_eros_e(uj_jelszo): hazirend_hiba = jelszo_hazirend.ellenorzes(db, target_user, uj_jelszo) # v36: a cég jelszó-házirendje if hazirend_hiba: return JSONResponse(status_code=400, content={"message": hazirend_hiba}) aktivalas_utils.kezi_aktivalas(db, target_user, current_user, uj_jelszo) elif not is_admin: aktivalas_utils.ujrakuldes(db, target_user, current_user) elif (is_admin or is_sajat_ceg_admin) and is_company_active(target_user.company_id, db) and not current_user.csak_olvasas and jelszo_eleg_eros_e(uj_jelszo): hazirend_hiba = jelszo_hazirend.ellenorzes(db, target_user, uj_jelszo) # v36: a cég jelszó-házirendje (bonyolultság, előzmény) if hazirend_hiba: return JSONResponse(status_code=400, content={"message": hazirend_hiba}) regi_hash = target_user.hashed_password target_user.hashed_password = get_password_hash(uj_jelszo) # JWT-visszavonás (2026-10-03-i biztonsági audit, 3. fázis): a jelszóváltás # időpontjának rögzítése - a security.py get_current_user() ezt veti össze a JWT # "iat" (kiállítás) mezőjével, így egy korábban kiadott (pl. ellopott) token a # jelszóváltással azonnal érvényét veszti, nem kell a lejáratáig (exp) kivárni. target_user.jelszo_valtozott_at = datetime.now(timezone.utc) jelszo_hazirend.uj_jelszo_utan(db, target_user, regi_hash) # v36: előzmény + új lejárat # KÉTLÉPCSŐS HITELESÍTÉS (2FA 5. fázis - elveszett eszköz, docs/KETLEPCSOS_HITELESITES.md 5e): a jelszócserével EGY lépésben, # ugyanabban a commitban törlődik a felhasználó 2FA-beállítása is (új jelszó + új QR). Kivétel: a SAJÁT beállítás és a # rendszeradmin beállítása (azokat egy ellopott munkamenet ne tudja lekapcsolni) - azokat a szerverről: mfa_felold.py. mfa_torolve = mfa_utils.visszaallitas_jelszocserevel(db, target_user, current_user) cel_id, vegrehajto_id = target_user.id, current_user.id # J-20 (v77): a jelszó cseréje naplózódik (ÉRTÉK nélkül - a jelszó / hash soha nem kerül a naplóba) naploz(db, current_user, target_user.company_id, "Jelszó módosítva (Admin által)", None, target_user.username) db.commit() if mfa_torolve: mfa_utils.ertesito_visszaallitasrol(db, cel_id, vegrehajto_id) referer_url = request.headers.get("referer") if not referer_url: referer_url = "/lista" return RedirectResponse(url=referer_url, status_code=status.HTTP_303_SEE_OTHER) @router.post("/sajat-jelszo-csere") async def sajat_jelszo_csere( request: Request, regi_jelszo: str = Form(...), uj_jelszo: str = Form(...), uj_jelszo2: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): # v27: az új jelszót MEGERŐSÍTENI kell (uj_jelszo2) - így egy elgépelt jelszóval nem zárhatja ki magát a felhasználó. # Szándékosan Form("") és nem Form(...): egy régi (gyorsítótárazott) oldal, amely még nem küldi a mezőt, érthető # hibaüzenetet kap, nem nyers 422-t. Az ellenőrzés a régi jelszó vizsgálata ELŐTT van (gyors, érthető visszajelzés). if uj_jelszo != uj_jelszo2: return JSONResponse(status_code=400, content={"message": "A két új jelszó nem egyezik!"}) jelszo_helyes = verify_password(regi_jelszo, current_user.hashed_password) if not jelszo_helyes: return JSONResponse(status_code=400, content={"message": "Hibás régi jelszó!"}) if not jelszo_eleg_eros_e(uj_jelszo): return JSONResponse(status_code=400, content={"message": "A jelszónak legalább 6 karakter hosszúnak kell lennie!"}) # v36: a cég jelszó-házirendje (bonyolultság, előzmény) - docs/JELSZO_HAZIREND.md hazirend_hiba = jelszo_hazirend.ellenorzes(db, current_user, uj_jelszo) if hazirend_hiba: return JSONResponse(status_code=400, content={"message": hazirend_hiba}) regi_hash = current_user.hashed_password current_user.hashed_password = get_password_hash(uj_jelszo) # JWT-visszavonás - l. a fenti admin_jelszo_csere() azonos megjegyzését. current_user.jelszo_valtozott_at = datetime.now(timezone.utc) jelszo_hazirend.uj_jelszo_utan(db, current_user, regi_hash) # v36: előzmény + új lejárat naploz(db, current_user, current_user.company_id, "Jelszó módosítva (saját)", None, current_user.username) # J-20 (v77): érték nélkül db.commit() return JSONResponse(status_code=200, content={"message": "Sikeres jelszócsere!"}) @router.post("/uj-ceges-felhasznalo") async def uj_ceges_felhasznalo( request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user) ): if current_user.role != "Admin": return JSONResponse(status_code=403, content={"message": "Nincs jogosultságod!"}) if current_user.csak_olvasas: return JSONResponse(status_code=403, content={"message": "Csak megtekintési jogosultsággal rendelkezel, szerkesztés nem lehetséges!"}) form_data = await request.form() username = form_data.get("username") email = form_data.get("email") role = form_data.get("role") if role not in ENGEDELYEZETT_UJ_FELHASZNALO_SZEREPKOROK + [TUZVEDELMI_KARBANTARTO_SZEREPKOR]: return JSONResponse(status_code=400, content={"message": "Érvénytelen szerepkör!"}) hozzaferes_szint = form_data.get("hozzaferes_szint") or "CEG" if hozzaferes_szint not in ENGEDELYEZETT_HOZZAFERES_SZINTEK: # Ismeretlen érték - a legszűkebb (GEP, alapból 0 hozzárendelt géppel) legyen az # alapértelmezett, NEM a legtágabb CEG, nehogy egy elgépelés/manipulált form-mező # véletlenül teljes céges hozzáférést eredményezzen. hozzaferes_szint = "GEP" working_company_id = get_working_company_id(current_user, request) if not is_company_active(working_company_id, db): return JSONResponse(status_code=403, content={"message": "Ez a cég inaktív státuszú, új felhasználó nem vehető fel hozzá!"}) if role == TUZVEDELMI_KARBANTARTO_SZEREPKOR: # J-18 (v66): Karbantartó csak Tűzvédelem-modulos cégnél értelmes (a szakvizsgához kötött karbantartás-rögzítés ott él) celceg = db.query(models.Company).filter(models.Company.id == working_company_id).first() if not celceg or "TUZVEDELEM" not in [m.kod for m in celceg.engedelyezett_modulok]: return JSONResponse(status_code=400, content={"message": "Karbantartó csak olyan cégnél vehető fel, amely előfizetett a Tűzvédelem modulra!"}) if not aktivalas_utils.email_ervenyes_e(email): return JSONResponse(status_code=400, content={"message": "Érvénytelen e-mail cím! Az aktiváló levelet erre a címre küldjük."}) letezo = db.query(models.User).filter(models.User.email == email).first() if letezo: return JSONResponse(status_code=400, content={"message": "Ez az e-mail cím már foglalt!"}) if role == "VIVU": van_mar_vivu = db.query(models.User).filter( models.User.company_id == working_company_id, models.User.role == "VIVU", models.User.aktiv == True, ).first() if van_mar_vivu: return JSONResponse(status_code=400, content={ "message": f'Ennél a cégnél már van aktív VIVÜ ügyintéző ("{van_mar_vivu.username}") - ' f'cégenként csak egy lehet. Előbb inaktiváld vagy módosítsd a meglévőt.' }) ceg = db.query(models.Company).filter(models.Company.id == working_company_id).first() tullepne, jelenlegi, limit = licenc_tullepne_uj_felhasznalonal(ceg, role, db) if tullepne: return JSONResponse( status_code=403, content={"message": f"Elérted a licenced szerinti maximális létszámot ehhez a szerepkörhöz ({jelenlegi}/{limit})! A bővítéshez keresd az üzemeltetőt."} ) # J-13 (v69): a telephely-/gép-korlátozás ellenőrzése a fájl-írás ELŐTT (hibánál 400, semmi sem keletkezik) uj_telephelyek, uj_gepek, hozzaferes_hiba = hozzaferes_alanyok_ellenorzese( db, working_company_id, hozzaferes_szint, form_data.getlist("allowed_sites"), form_data.getlist("allowed_machines")) if hozzaferes_hiba: return JSONResponse(status_code=400, content={"message": hozzaferes_hiba}) file_path = None bizonyitvany_file = form_data.get("bizonyitvany_file") if bizonyitvany_file: if getattr(bizonyitvany_file, "filename", "") != "": if not is_safe_extension(bizonyitvany_file.filename): return JSONResponse(status_code=400, content={"message": "Nem engedélyezett fájlformátum a bizonyítványnál!"}) _fejresz = bizonyitvany_file.file.read(8) bizonyitvany_file.file.seek(0) if not tartalom_illeszkedik_e_kiterjesztesre(bizonyitvany_file.filename, _fejresz): return JSONResponse(status_code=400, content={"message": "A feltöltött fájl tartalma nem egyezik a kiterjesztésével!"}) mappa = f"uploads/{working_company_id}/bizonyitvanyok" os.makedirs(mappa, exist_ok=True) safe_email = sanitize_filename(email) safe_filename = sanitize_filename(bizonyitvany_file.filename) file_path = f"/{mappa}/{safe_email}_{safe_filename}" with open(file_path.lstrip('/'), "wb+") as f: shutil.copyfileobj(bizonyitvany_file.file, f) jog_uzem = False if form_data.get("jog_uzemora_rogzites") == "on": jog_uzem = True jog_dok = False if form_data.get("jog_dokumentum_feltoltes") == "on": jog_dok = True jog_datum = False if form_data.get("jog_datum_szerkesztes") == "on": jog_datum = True # --- ÚJ: HR DÁTUMOK BEOLVASÁSA --- try: orvosi_datum = None if form_data.get("orvosi_alkalmassag_ervenyesseg"): orvosi_datum = datetime.strptime(form_data.get("orvosi_alkalmassag_ervenyesseg"), "%Y-%m-%d").date() jogositvany_datum = None if form_data.get("jogositvany_ervenyesseg"): jogositvany_datum = datetime.strptime(form_data.get("jogositvany_ervenyesseg"), "%Y-%m-%d").date() except ValueError: return JSONResponse(status_code=400, content={"message": "Érvénytelen dátumformátum!"}) # --------------------------------- uj_user = models.User( username=username, email=email, # AKTIVÁLÁS (E szelet): nincs kezdőjelszó - a felhasználó az aktiváló linken állítja be (aktivalas_utils.py) hashed_password=aktivalas_utils.jelszo_nelkuli_hash(), role=role, company_id=working_company_id, hozzaferes_szint=hozzaferes_szint, bizonyitvany_szam=form_data.get("bizonyitvany_szam", ""), bizonyitvany_path=file_path, jog_uzemora_rogzites=jog_uzem, jog_dokumentum_feltoltes=jog_dok, jog_datum_szerkesztes=jog_datum, orvosi_alkalmassag_ervenyesseg=orvosi_datum, jogositvany_ervenyesseg=jogositvany_datum, csak_olvasas=(form_data.get("csak_olvasas") == "on") ) uj_user.allowed_sites.extend(uj_telephelyek) # J-13 (v69): már ellenőrzött, a cégre szűrt listák uj_user.allowed_machines.extend(uj_gepek) # A cég ÖSSZES előfizetett modulját megkapja - enélkül az új felhasználó a bejelentkezés # után rögtön elakadna a modulválasztó oldalon (0 modul), egészen a következő szerver- # újraindításig (ami egy háttér-migrációval pótolná ezt a hiányzó hozzárendelést). # KIVÉTEL: a "VIVU" szerepkör (és ugyanígy a routers/vivu.py-ban felvett "Megtekintő") egy # szűken körülhatárolt feladatkör - CSAK a VIVÜ modulhoz kell hozzáférnie, még akkor is, ha # a cégnek egyébként más modulja (pl. Emelőgép) is elő van fizetve, különben a VIVÜ # ügyintéző bejelentkezés után feleslegesen (és jogosultsági szempontból indokolatlanul) # egy modulválasztó képernyőt látna az Emelőgép-nyilvántartással is. if role == "VIVU": vivu_modul = db.query(models.Modul).filter(models.Modul.kod == "VIVU").first() uj_user.sajat_modulok = [vivu_modul] if vivu_modul else [] elif role == TUZVEDELMI_KARBANTARTO_SZEREPKOR: # J-18 (v66): csak a Tűzvédelem modul - az Emelőgép modul szerepkör-ellenőrzései (J-25: tiltó listák) a Karbantartót átengednék tuz_modul = db.query(models.Modul).filter(models.Modul.kod == "TUZVEDELEM").first() uj_user.sajat_modulok = [tuz_modul] if tuz_modul else [] else: uj_user.sajat_modulok = list(ceg.engedelyezett_modulok) if ceg else [] db.add(uj_user) aktivalas_utils.meghivas(db, uj_user, current_user) # aktiváló levél (E szelet) # ÚJ (2026-10-02-i user-jelzés): a felhasználó-létrehozás eddig egyáltalán nem került be a # Jogosultsági naplóba - csak az aktiválás/inaktiválás (l. kapcsol_felhasznalo_aktiv). db.add(models.AuditLog( company_id=working_company_id, user_id=current_user.id, field_name="Új felhasználó", old_value="-", new_value=f"{username} ({email}) - szerepkör: {role}" )) db.commit() referer_url = request.headers.get("referer") if not referer_url: referer_url = "/beallitasok" return RedirectResponse(url=referer_url, status_code=status.HTTP_303_SEE_OTHER) # A cégek (SaaS bérlők) teljes menedzsmentje (létrehozás, aktiválás, végleges törlés) mostantól # a rendszeradminisztrátori felületen (routers/rendszeradmin.py) él, nem itt.