"""Rendszeradmin: teljes felhasználókezelés a /felhasznalok oldalról (v25: szerepkör/cég/extra cégek; v26: MINDEN más művelet). A modul két részből áll: 1. rész (v25): szerepkör / cég / extra cégek módosítása - POST /felhasznalok/{id}/szerep-ceg (lásd alább). 2. rész (v26, "3. szelet"): a rendszeradminisztrátor MINDENT megtehet BÁRMELYIK felhasználóval (a többi rendszeradminnal is), a "Kezelés" ablakból: teljes szerkesztés (minden paraméter, modulok, telephely/gép), jelszó, 2FA törlés/visszaállítás, aktív/inaktív, végleges törlés. Lásd a fájl második felét ("3. szelet") és docs/FELHASZNALOK_LISTA.md 17-23. fejezet. --- 1. rész: szerepkör / cég / extra cégek (v25) --- Rendszeradmin: egy felhasználó SZEREPKÖRÉNEK, CÉGÉNEK és EXTRA CÉGEINEK módosítása (v25, 2. szelet). Leírás részletesen: docs/FELHASZNALOK_LISTA.md (a "2. szelet" fejezetek). VÉGPONT: POST /felhasznalok/{user_id}/szerep-ceg (a /felhasznalok oldal "Szerepkör / cég" gombja és modálja hívja) MIÉRT KELL? Eddig SEMMI nem módosíthatta egy felhasználó szerepkörét vagy cégét a létrehozás után (a /szerkeszt-felhasznalo tudatosan csak a nevet, e-mailt és a jelzőket állítja). Ha valaki rossz szerepkörrel vagy rossz cégben jött létre, törölni és újra felvenni kellett. Ez a végpont ezt oldja meg - ÉS ez a rendszer legveszélyesebb felhasználó-művelete (a szerepkör és a cég dönti el, ki mit láthat), ezért a lehető legszigorúbb védőkorlátok mögött van. JOGOSULTSÁG: KIZÁRÓLAG a rendszeradminisztrátor (Admin + company_id == 1). Cég-Admin és mindenki más: 403, semmi sem változik. VÉDŐKORLÁTOK (mindegyiket a tests/t_userlist2.py ellenőrzi): 1. csak rendszeradmin, csak "csak olvasás" jelző nélkül, csak a saját oldalról érkező kérés (Sec-Fetch-Site: CSRF elleni mélységi védelem); 2. a SAJÁT fiókot nem lehet így módosítani (az RA ne tudja véletlenül lefokozni/elköltöztetni magát); 3. az új szerepkör a 9 ismert közül való (a régi "Szuperfelhasználó" stb. NEM választható); 4. az új cég LÉTEZIK és AKTÍV, és NEM a rendszercég (1): ide semmilyen úton nem lehet RA-t csinálni (Admin + 1-es cég tiltott; új rendszeradmint továbbra is csak a szerveren futtatott szkript hozhat létre - lásd docs/JOGOSULTSAGOK.md 3.5); 5. RA fiók visszaminősítése/áthelyezése: csak ha marad másik AKTÍV RA (az "utolsó RA" nem fokozható le), ÉS be kell gépelni a fiók nevét (külön megerősítés a szokásos jelölőnégyzeten felül); 6. minden módosításhoz be kell jelölni a megerősítést; 7. a szerepkörhöz tartozó szabályok érvényesülnek: cégenként egyetlen aktív VIVU, licenc-limit (Felhasználó/EÜI), a VIVU/EBK szerepkörhöz a cégnek elő kell fizetnie a modult; 8. a modulok a szerepkörhöz/céghez igazodnak (modulok_igazitasa), a régi cég hozzárendelései (telephely, gép, EBK/VIVÜ megbízások, utolsó munkakörnyezet) törlődnek, hogy a felhasználó ne vigyen át jogosultságot az új cégbe; 9. minden változás AuditLog bejegyzést kap (a Jogosultsági naplóban látszik), kérésből érkező szöveg nem kerül a válaszba. Az egész művelet egyetlen tranzakció: bármelyik ellenőrzés elbukik -> semmi nem módosul. """ from fastapi import APIRouter, Request, Depends, Form from fastapi.responses import JSONResponse from sqlalchemy import delete from sqlalchemy.orm import Session import models from database import get_db from licenc_utils import licenc_tullepne_uj_felhasznalonal, licenc_reaktivalas_hiba from routers.rendszeradmin_mfa import _idegen_eredet # ugyanaz a Sec-Fetch-Site CSRF-védelem, mint a 2FA-kapcsolónál from security import get_current_user, is_rendszeradminisztrator router = APIRouter() # A rendszer 9 ismert szerepköre (docs/JOGOSULTSAGOK.md 1. fejezet). Az "Admin" itt CSAK cég-Adminként választható (nem az 1-es cégben). SZEREPKOROK = [ "Admin", "Emelőgép ügyintéző", "Emelőgép szakértő", "Emelőgép vizsgáló", "Felhasználó", "Karbantartó", "VIVU", "EBK", "Megtekintő", ] # Azok a szerepkörök, amelyeknél az "extra cégek" (allowed_companies) értelmezett (ugyanaz, mint az /uj-admin-felhasznalo-ban). KULSOS_SZAKERTOI_SZEREPKOROK = ("Emelőgép szakértő", "Emelőgép vizsgáló") # Ezek a szerepkörök a cég ÖSSZES előfizetett modulját megkapják (ahogy a létrehozáskor is). ALTALANOS_SZEREPKOROK = ("Admin", "Emelőgép ügyintéző", "Emelőgép szakértő", "Emelőgép vizsgáló", "Felhasználó", "Karbantartó") def modulok_igazitasa(uj_ceg, uj_szerep: str, regi_szerep: str, regi_modulok: list): """Az új szerepkörhöz/céghez illő modulok. Visszaad: (modul_lista, hibaüzenet|None). Szabályok (a létrehozó végpontok mintája szerint): * VIVU -> csak a VIVU modul; EBK -> csak az EBK modul (ha a cég nem fizette elő: HIBA, nem találgatunk); * Megtekintő -> a RÉGI moduljai a cég előfizetésével metszve (a Megtekintő vagy az EBK, vagy a VIVÜ modult használja, ezt a szerepkör nem dönti el - ezért ha nincs mit megtartani: HIBA, a modult előbb a Rendszeradminisztráció/Modulok oldalon kell kiosztani); * minden más szerepkör -> ha eddig is "általános" szerepkörű volt és voltak moduljai: a régiek ∩ a cég modulai (a kézzel szűkített hozzárendelés nem bővül); különben a cég összes aktív modulja. """ ceg_modulok = [m for m in uj_ceg.engedelyezett_modulok if m.aktiv] ceg_kodok = {m.kod: m for m in ceg_modulok} if uj_szerep in ("VIVU", "EBK"): if uj_szerep not in ceg_kodok: return None, f'A cég nem rendelkezik a "{uj_szerep}" szerepkörhöz szükséges {uj_szerep} modullal - előbb a cég előfizetéséhez kell adni.' return [ceg_kodok[uj_szerep]], None regi_ceg_metszet = [m for m in regi_modulok if m.aktiv and m.kod in ceg_kodok] if uj_szerep == "Megtekintő": if not regi_ceg_metszet: return None, ('A "Megtekintő" szerepkörhöz a felhasználónak EBK vagy VIVU modulra van szüksége, de jelenleg nincs olyan modulja, ' 'amit az új cég is előfizetett. Előbb oszd ki a modult, vagy válassz másik szerepkört.') return regi_ceg_metszet, None if regi_szerep in ALTALANOS_SZEREPKOROK and regi_ceg_metszet: return regi_ceg_metszet, None return list(ceg_modulok), None def _hiba(kod: int, uzenet: str) -> JSONResponse: return JSONResponse(status_code=kod, content={"message": uzenet}) def _naplo(db: Session, ceg_id, actor, mezo: str, regi: str, uj: str): db.add(models.AuditLog(company_id=ceg_id, user_id=actor.id, field_name=mezo, old_value=regi, new_value=uj)) def szerep_ceg_modositas(db: Session, actor: models.User, user_id: int, uj_szerep: str, uj_ceg_id, extra_ceg_idk: list, megerosites: bool, nev_megerosites: str): """A művelet magja (a végpont és a teszt is ezt hívja). Visszaad: (http_kod, {"message": ..., ["valtozasok": [...]]}). A JOGOSULTSÁGOT (RA-kapu, CSRF) a végpont ellenőrzi, de a "csak olvasás" és a saját-fiók védelem itt is megvan (mélységi védelem).""" if not is_rendszeradminisztrator(actor): return 403, {"message": "Ehhez a művelethez rendszeradminisztrátori jogosultság kell."} if actor.csak_olvasas: return 403, {"message": "Csak megtekintési jogosultsággal rendelkezel, a módosítás nem lehetséges."} cel = db.query(models.User).filter(models.User.id == user_id).first() if not cel: return 404, {"message": "A felhasználó nem található."} if cel.id == actor.id: return 403, {"message": "A saját fiókod szerepköre és cége nem módosítható (különben véletlenül kizárhatnád magad)."} if not megerosites: return 400, {"message": "A módosításhoz be kell jelölni a megerősítést."} if uj_szerep not in SZEREPKOROK: return 400, {"message": "Ismeretlen szerepkör."} try: uj_ceg_id = int(uj_ceg_id) except (TypeError, ValueError): return 400, {"message": "Érvénytelen cég."} if uj_ceg_id == 1: return 400, {"message": "A rendszercégbe (és ezzel a rendszeradminisztrátori szerepbe) innen nem lehet felhasználót tenni. " "Új rendszeradmin létrehozása csak a szerveren futtatott szkripttel lehetséges."} uj_ceg = db.query(models.Company).filter(models.Company.id == uj_ceg_id).first() if not uj_ceg: return 400, {"message": "A kiválasztott cég nem létezik."} if not uj_ceg.aktiv: return 400, {"message": "A kiválasztott cég inaktív, ide nem lehet felhasználót tenni."} if uj_szerep == "Admin" and uj_ceg_id == 1: # (az előző ág már kizárja; a szabály így külön, kifejezetten is rögzített) return 400, {"message": "Rendszeradmin ezzel a művelettel nem hozható létre."} regi_szerep, regi_ceg_id = cel.role, cel.company_id regi_ceg = db.query(models.Company).filter(models.Company.id == regi_ceg_id).first() ra_cel = is_rendszeradminisztrator(cel) # --- RA fiók visszaminősítése: csak ha marad másik aktív RA + a név begépelése if ra_cel: marad = db.query(models.User).filter( models.User.role == "Admin", models.User.company_id == 1, models.User.aktiv.is_(True), models.User.id != cel.id ).all() marad = [u for u in marad if u.aktivalva is not False] if not marad: return 403, {"message": "Ez az utolsó aktív rendszeradminisztrátor, nem fokozható le."} if (nev_megerosites or "").strip() != (cel.username or "").strip(): return 400, {"message": "Rendszeradminisztrátor visszaminősítéséhez a megerősítő mezőbe pontosan be kell gépelni a fiók nevét."} szerep_valt = uj_szerep != regi_szerep ceg_valt = uj_ceg_id != regi_ceg_id # --- extra cégek (csak szakértő/vizsgáló): létező, aktív, nem rendszercég, nem az új saját cég uj_extra = [] if uj_szerep in KULSOS_SZAKERTOI_SZEREPKOROK: for raw in extra_ceg_idk: try: cid = int(raw) except (TypeError, ValueError): return 400, {"message": "Érvénytelen extra cég azonosító."} if cid == 1 or cid == uj_ceg_id or any(c.id == cid for c in uj_extra): continue c = db.query(models.Company).filter(models.Company.id == cid).first() if not c or not c.aktiv: return 400, {"message": "Az extra cégek között nem létező vagy inaktív cég van."} uj_extra.append(c) regi_extra = list(cel.allowed_companies) extra_valt = {c.id for c in uj_extra} != {c.id for c in regi_extra} if not (szerep_valt or ceg_valt or extra_valt): return 200, {"message": "Nem volt változás.", "valtozasok": []} # --- szerepkör-szabályok az ÚJ cégben if cel.aktiv and (szerep_valt or ceg_valt): if uj_szerep == "VIVU": masik = db.query(models.User).filter( models.User.company_id == uj_ceg_id, models.User.role == "VIVU", models.User.aktiv.is_(True), models.User.id != cel.id ).first() if masik: return 400, {"message": f'Az új cégnél már van aktív VIVÜ ügyintéző ("{masik.username}") - cégenként csak egy lehet.'} tullepne, jelenlegi, limit = licenc_tullepne_uj_felhasznalonal(uj_ceg, uj_szerep, db) if tullepne: return 400, {"message": f"A cég licenc-limitje betelt ({jelenlegi}/{limit}) ennél a szerepkörnél, a felhasználó nem tehető ide."} uj_modulok, modul_hiba = modulok_igazitasa(uj_ceg, uj_szerep, regi_szerep, list(cel.sajat_modulok)) if modul_hiba: return 400, {"message": modul_hiba} # ===== MINDEN ELLENŐRZÉS RENDBEN - a módosítás ===== valtozasok = [] nev = cel.username regi_modul_kodok = sorted(m.kod for m in cel.sajat_modulok) if szerep_valt: cel.role = uj_szerep valtozasok.append(f"szerepkör: {regi_szerep} -> {uj_szerep}") for ceg_id in {regi_ceg_id, uj_ceg_id}: _naplo(db, ceg_id, actor, "Szerepkör módosítása (rendszeradmin)", f"{nev}: {regi_szerep}", f"{nev}: {uj_szerep}") if ceg_valt: cel.company_id = uj_ceg_id regi_nev = f"{regi_ceg.nev} (#{regi_ceg_id})" if regi_ceg else f"#{regi_ceg_id}" valtozasok.append(f"cég: {regi_nev} -> {uj_ceg.nev} (#{uj_ceg_id})") for ceg_id in {regi_ceg_id, uj_ceg_id}: _naplo(db, ceg_id, actor, "Cég módosítása (rendszeradmin)", f"{nev}: {regi_nev}", f"{nev}: {uj_ceg.nev} (#{uj_ceg_id})") # A régi cég hozzárendelései NEM vihetők át (más cég telephelyei/gépei/projektjei): # a hozzáférési SZINT (CEG/TELEPHELY/GEP) szándékosan megmarad - a szűkített felhasználó így az új cégben is szűkített # (nulla találat) marad, amíg az új cég Admin-ja ki nem jelöl neki telephelyet/gépet. A szint "szélesítése" TILOS. torolt = len(cel.allowed_sites) + len(cel.allowed_machines) cel.allowed_sites = [] cel.allowed_machines = [] cel.telephely_hozzaferes_id = None cel.utolso_ceg_id = None if torolt: _naplo(db, uj_ceg_id, actor, "Hozzáférés-hozzárendelések törlése (cégváltás)", f"{nev}: {torolt} telephely/gép", f"{nev}: 0") if szerep_valt or ceg_valt: # EBK/VIVÜ megbízások (Megtekintő/EBK szerepkör szűkítése): a régi szerep/cég projektjei/ügyfelei nem maradhatnak for tabla, mezo in ((models.user_ebk_projektek, "user_id"), (models.ebk_projekt_koordinatorok, "user_id"), (models.user_vivu_ugyfelek, "user_id")): db.execute(delete(tabla).where(getattr(tabla.c, mezo) == cel.id)) db.expire(cel, ["ebk_projektek", "ebk_koordinalt_projektek", "vivu_ugyfelek"]) cel.sajat_modulok = uj_modulok uj_modul_kodok = sorted(m.kod for m in uj_modulok) if uj_modul_kodok != regi_modul_kodok: valtozasok.append(f"modulok: {', '.join(regi_modul_kodok) or '-'} -> {', '.join(uj_modul_kodok) or '-'}") _naplo(db, uj_ceg_id, actor, "Modulok igazítása (rendszeradmin)", f"{nev}: {', '.join(regi_modul_kodok) or '-'}", f"{nev}: {', '.join(uj_modul_kodok) or '-'}") if extra_valt or szerep_valt or ceg_valt: # (szerepkörváltásnál az extra cégek is újraértékelődnek: csak szakértő/vizsgálónál maradhatnak) regi_kod = sorted(c.id for c in regi_extra) uj_kod = sorted(c.id for c in uj_extra) if regi_kod != uj_kod: cel.allowed_companies = uj_extra valtozasok.append(f"extra cégek: {regi_kod or '-'} -> {uj_kod or '-'}") _naplo(db, uj_ceg_id, actor, "Extra cégek módosítása (rendszeradmin)", f"{nev}: {', '.join(str(i) for i in regi_kod) or '-'}", f"{nev}: {', '.join(str(i) for i in uj_kod) or '-'}") eltavolitott = {c.id for c in regi_extra} - {c.id for c in uj_extra} if eltavolitott: # a szakértői megbízás-beállítások (vizsgálat-típus szűrés) az elvett cégekhez ne maradjanak db.query(models.SzakertoMegbizatas).filter( models.SzakertoMegbizatas.user_id == cel.id, models.SzakertoMegbizatas.company_id.in_(eltavolitott) ).delete(synchronize_session=False) db.commit() return 200, {"message": "A módosítás megtörtént: " + "; ".join(valtozasok) + ".", "valtozasok": valtozasok} @router.post("/felhasznalok/{user_id}/szerep-ceg") async def szerep_ceg_modositas_vegpont( user_id: int, request: Request, role: str = Form(...), company_id: str = Form(...), megerosites: str = Form(""), nev_megerosites: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user), ): # 1. JOGOSULTSÁG: csak a rendszeradminisztrátor (mindenki más: 403, semmi nem módosul, és az üzenet sem árul el semmit) if not is_rendszeradminisztrator(current_user): return _hiba(403, "Ehhez a művelethez rendszeradminisztrátori jogosultság kell.") # 2. CSRF mélységi védelem (lásd routers/rendszeradmin_mfa.py _idegen_eredet) if _idegen_eredet(request): return _hiba(403, "A kérés nem erről az oldalról érkezett, ezért nem hajtottuk végre.") form = await request.form() kod, tartalom = szerep_ceg_modositas( db, current_user, user_id, role, company_id, form.getlist("allowed_companies"), megerosites in ("on", "1", "true"), nev_megerosites, ) return JSONResponse(status_code=kod, content=tartalom) # ============================================================================================================================= # 3. SZELET (v26): a rendszeradminisztrátor MINDENT megtehet BÁRMELYIK felhasználóval # ============================================================================================================================= # Megrendelői döntés (2026-10-05): "a rendszeradmin MINDENT megtehet a userrel: törölheti, szerkesztheti, a 2FA-t menedzselheti (törölheti, # visszaállíthatja - egyelőre), jelszót resetelhet". Ez a más RENDSZERADMINISZTRÁTOR fiókokra is vonatkozik. # # Ami KORLÁT marad (és miért) - ezeket NEM a "jogosultság" miatt tartjuk meg, hanem a kizárás/elvesztés megelőzésére: # * a SAJÁT fiókot nem lehet inaktiválni / törölni / "csak olvasás"-ra állítani / szerepkörét-cégét módosítani, és a saját jelszó/2FA # nem ezen a felületen változik (a "Jelszavam módosítása" és a /rendszeradmin/ketlepcsos oldal a megfelelő hely); # * az UTOLSÓ aktív rendszeradmin nem inaktiválható/törölhető/fokozható le; # * másik RENDSZERADMIN jelszavának, 2FA-jának törléséhez és a fiók törléséhez a végrehajtónak be kell írnia a SAJÁT jelszavát (újra- # azonosítás). Ez nem korlátoz jogosultságot, csak azt akadályozza meg, hogy egy ELLOPOTT/őrizetlenül hagyott rendszeradmin- # munkamenet egyetlen kattintással átvehesse a többi rendszeradmin fiókját (a 2FA-védelem eddig pont ezt a kört zárta ki - lásd # mfa_utils.visszaallitas_jelszocserevel). A másik rendszeradminok emailes MÁSOLATOT kapnak minden ilyen műveletről. # Minden művelet: csak RA, idegen eredet (cross-site) tiltva, "csak olvasás" jelzős RA nem módosíthat, AuditLog bejegyzés. from datetime import datetime, timezone import os import shutil import aktivalas_utils import jelszo_hazirend import mfa_utils from security import ( get_password_hash, is_safe_extension, jelszo_eleg_eros_e, sanitize_filename, tartalom_illeszkedik_e_kiterjesztesre, verify_password, ) HOZZAFERES_SZINTEK = ("CEG", "TELEPHELY", "GEP") def _kapu(request: Request, actor: models.User, db: Session, user_id: int, modosit: bool = True): """Közös kapu minden 3. szelet végponthoz. Visszaad: (cel, None) vagy (None, JSONResponse hiba).""" if not is_rendszeradminisztrator(actor): return None, _hiba(403, "Ehhez a művelethez rendszeradminisztrátori jogosultság kell.") if modosit: if _idegen_eredet(request): return None, _hiba(403, "A kérés nem erről az oldalról érkezett, ezért nem hajtottuk végre.") if actor.csak_olvasas: return None, _hiba(403, "Csak megtekintési jogosultsággal rendelkezel, a módosítás nem lehetséges.") cel = db.query(models.User).filter(models.User.id == user_id).first() if not cel: return None, _hiba(404, "A felhasználó nem található.") return cel, None def _marad_masik_aktiv_ra(db: Session, cel: models.User) -> bool: """Marad-e legalább egy MÁSIK aktív (és már aktivált) rendszeradmin, ha a `cel` kiesik? (az "utolsó RA" védelem)""" masok = db.query(models.User).filter( models.User.role == "Admin", models.User.company_id == 1, models.User.aktiv.is_(True), models.User.id != cel.id ).all() return any(u.aktivalva is not False for u in masok) def _ujraazonositas_kell(cel: models.User) -> bool: """Másik rendszeradmin fiókján végzett érzékeny műveletnél a végrehajtónak újra azonosítania kell magát a saját jelszavával.""" return is_rendszeradminisztrator(cel) def _ujraazonositas_rendben(actor: models.User, sajat_jelszo: str) -> bool: try: return bool(sajat_jelszo) and verify_password(sajat_jelszo, actor.hashed_password) except Exception: return False def _ertek_szoveg(v) -> str: if v is None or v == "": return "-" if isinstance(v, bool): return "igen" if v else "nem" return str(v) # --------------------------------------------------------------------------------------------------------------------------- # A "Kezelés" ablak adatai (csak olvasás): GET /felhasznalok/{id}/adatok # --------------------------------------------------------------------------------------------------------------------------- @router.get("/felhasznalok/{user_id}/adatok") def felhasznalo_adatok(user_id: int, request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): """A Kezelés ablak kitöltéséhez: a felhasználó MINDEN szerkeszthető paramétere + a választékok (a cég telephelyei/gépei/modulai). SOHA nem tartalmaz titkot (jelszó-hash, TOTP-titok). A 2FA-ról csak az állapot-szöveg szerepel.""" cel, hiba = _kapu(request, current_user, db, user_id, modosit=False) if hiba: return hiba from routers.rendszeradmin import _felhasznalo_torlesi_szamlalo # késői import: a rendszeradmin.py sok mindent betölt ceg = cel.ceg cegmodulok = [m for m in (ceg.engedelyezett_modulok if ceg else [])] datum = lambda d: d.strftime("%Y-%m-%d") if d else "" # noqa: E731 return JSONResponse(content={ "id": cel.id, "username": cel.username, "email": cel.email, "role": cel.role, "company_id": cel.company_id, "company_name": ceg.nev if ceg else "", "aktiv": bool(cel.aktiv), "aktivalva": cel.aktivalva is not False, "csak_olvasas": bool(cel.csak_olvasas), "ra": is_rendszeradminisztrator(cel), "sajat": cel.id == current_user.id, "ujraazonositas": _ujraazonositas_kell(cel), "hozzaferes_szint": cel.hozzaferes_szint or "CEG", "allowed_sites": [s.id for s in cel.allowed_sites], "allowed_machines": [m.id for m in cel.allowed_machines], "sites": [{"id": s.id, "nev": s.nev} for s in db.query(models.Site).filter(models.Site.company_id == cel.company_id).order_by(models.Site.nev).all()], "machines": [{"id": m.id, "nev": f"{m.megnevezes or '?'} ({m.gyari_szam or '-'})"} for m in db.query(models.Machine).filter(models.Machine.company_id == cel.company_id).order_by(models.Machine.megnevezes).limit(5000).all()], "jog_uzemora_rogzites": bool(cel.jog_uzemora_rogzites), "jog_dokumentum_feltoltes": bool(cel.jog_dokumentum_feltoltes), "jog_gep_szerkesztes": bool(cel.jog_gep_szerkesztes), "jog_datum_szerkesztes": bool(cel.jog_datum_szerkesztes), "bizonyitvany_szam": cel.bizonyitvany_szam or "", "kamarai_azonosito": cel.kamarai_azonosito or "", "engedely_szam": cel.engedely_szam or "", "tanusitvany_szam": cel.tanusitvany_szam or "", "bizonyitvany_path": cel.bizonyitvany_path or "", "orvosi_alkalmassag_ervenyesseg": datum(cel.orvosi_alkalmassag_ervenyesseg), "jogositvany_ervenyesseg": datum(cel.jogositvany_ervenyesseg), "ceg_modulok": [{"id": m.id, "kod": m.kod, "nev": m.nev} for m in cegmodulok if m.aktiv], "modulok": [m.id for m in cel.sajat_modulok], "mfa": {"allapot": mfa_utils.allapot(db, cel), "beallitva": mfa_utils.beallitasi_allapot(cel)}, "torlesi_osszegzes": _felhasznalo_torlesi_szamlalo(cel.id, db), }) # --------------------------------------------------------------------------------------------------------------------------- # TELJES SZERKESZTÉS: POST /felhasznalok/{id}/teljes-szerkesztes # --------------------------------------------------------------------------------------------------------------------------- def _fel_azonosito_lista(raw: list): ki = [] for r in raw: try: ki.append(int(r)) except (TypeError, ValueError): return None return ki @router.post("/felhasznalok/{user_id}/teljes-szerkesztes") async def teljes_szerkesztes(user_id: int, request: Request, db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): """A felhasználó MINDEN paraméterének szerkesztése (név, e-mail, aktív, csak olvasás, hozzáférési szint + telephelyek/gépek, jog_* jelzők, szakmai és HR adatok, bizonyítvány-fájl, modulok). A szerepkör/cég/extra cégek a /szerep-ceg végponton, a jelszó/2FA/törlés külön végpontokon változik. Minden VÁLTOZÁS mezőnként AuditLog bejegyzést kap. Egyetlen tranzakció: bármi hibás -> semmi nem módosul.""" cel, hiba = _kapu(request, current_user, db, user_id) if hiba: return hiba form = await request.form() sajat = cel.id == current_user.id ra_cel = is_rendszeradminisztrator(cel) valtozasok = [] # (mező, régi, új) def valt(mezo, regi, uj): if _ertek_szoveg(regi) != _ertek_szoveg(uj): valtozasok.append((mezo, _ertek_szoveg(regi), _ertek_szoveg(uj))) # --- név, e-mail nev = (form.get("username") or "").strip() email = (form.get("email") or "").strip() if not nev or len(nev) > 255: return _hiba(400, "A név megadása kötelező (legfeljebb 255 karakter).") regi_email = cel.email if email != cel.email: if not aktivalas_utils.email_ervenyes_e(email): return _hiba(400, "Az e-mail cím formátuma érvénytelen.") if db.query(models.User).filter(models.User.email == email, models.User.id != cel.id).first(): return _hiba(400, "Ez az e-mail cím már foglalt.") # --- állapot / csak olvasás uj_aktiv = form.get("aktiv") == "on" uj_csak = form.get("csak_olvasas") == "on" if sajat and (not uj_aktiv or uj_csak): return _hiba(403, "A saját fiókodat nem inaktiválhatod, és nem állíthatod csak olvasásra (kizárnád magad).") if ra_cel and (cel.aktiv and not uj_aktiv) and not _marad_masik_aktiv_ra(db, cel): return _hiba(403, "Ez az utolsó aktív rendszeradminisztrátor, nem inaktiválható.") if ra_cel and uj_csak and not cel.csak_olvasas and not _marad_masik_aktiv_ra(db, cel): return _hiba(403, "Ez az utolsó aktív rendszeradminisztrátor, nem állítható csak olvasásra.") if cel.role == "VIVU" and uj_aktiv and not cel.aktiv: masik = db.query(models.User).filter(models.User.company_id == cel.company_id, models.User.role == "VIVU", models.User.aktiv.is_(True), models.User.id != cel.id).first() if masik: return _hiba(400, f'Ennél a cégnél már van aktív VIVÜ ügyintéző ("{masik.username}") - cégenként csak egy lehet.') if uj_aktiv and not cel.aktiv: licenc_hiba = licenc_reaktivalas_hiba(cel, db) # J-06 (v66): a licenc-limit visszakapcsoláskor is él (a szerkesztő 'aktív' pipája is megkerülte) if licenc_hiba: return _hiba(400, licenc_hiba) # --- hozzáférési szint, telephelyek, gépek (CSAK a felhasználó cégének telephelyei/gépei - a régi szerkesztő ezt nem ellenőrizte) szint = form.get("hozzaferes_szint") or "CEG" if szint not in HOZZAFERES_SZINTEK: return _hiba(400, "Érvénytelen hozzáférési szint.") th_idk = _fel_azonosito_lista(form.getlist("allowed_sites")) gep_idk = _fel_azonosito_lista(form.getlist("allowed_machines")) if th_idk is None or gep_idk is None: return _hiba(400, "Érvénytelen telephely/gép azonosító.") uj_telephelyek, uj_gepek = [], [] if szint == "TELEPHELY": uj_telephelyek = db.query(models.Site).filter(models.Site.id.in_(th_idk), models.Site.company_id == cel.company_id).all() if th_idk else [] if len(uj_telephelyek) != len(set(th_idk)): return _hiba(400, "A telephelyek között van olyan, ami nem ennek a felhasználónak a cégéhez tartozik.") elif szint == "GEP": uj_gepek = db.query(models.Machine).filter(models.Machine.id.in_(gep_idk), models.Machine.company_id == cel.company_id).all() if gep_idk else [] if len(uj_gepek) != len(set(gep_idk)): return _hiba(400, "A gépek között van olyan, ami nem ennek a felhasználónak a cégéhez tartozik.") # --- modulok (csak a cég előfizetésén belüli aktív modulok) mod_idk = _fel_azonosito_lista(form.getlist("modulok")) if mod_idk is None: return _hiba(400, "Érvénytelen modul azonosító.") ceg_modul_idk = {m.id for m in (cel.ceg.engedelyezett_modulok if cel.ceg else []) if m.aktiv} if not set(mod_idk) <= ceg_modul_idk: return _hiba(400, "A felhasználónak csak olyan modul adható, amire a cége elő van fizetve.") uj_modulok = db.query(models.Modul).filter(models.Modul.id.in_(mod_idk)).all() if mod_idk else [] # --- dátumok def datum(kulcs): v = (form.get(kulcs) or "").strip() return datetime.strptime(v, "%Y-%m-%d").date() if v else None try: orvosi, jogositvany = datum("orvosi_alkalmassag_ervenyesseg"), datum("jogositvany_ervenyesseg") except ValueError: return _hiba(400, "Érvénytelen dátumformátum.") # --- bizonyítvány-fájl (ugyanazok az ellenőrzések, mint a létrehozásnál/régi szerkesztőnél) uj_fajl_path = None fajl = form.get("bizonyitvany_file") if fajl is not None and getattr(fajl, "filename", ""): if not is_safe_extension(fajl.filename): return _hiba(400, "Nem engedélyezett fájlformátum a bizonyítványnál.") fejresz = fajl.file.read(8); fajl.file.seek(0) if not tartalom_illeszkedik_e_kiterjesztesre(fajl.filename, fejresz): return _hiba(400, "A feltöltött fájl tartalma nem egyezik a kiterjesztésével.") mappa = f"uploads/{cel.company_id}/bizonyitvanyok" os.makedirs(mappa, exist_ok=True) uj_fajl_path = f"/{mappa}/{sanitize_filename(email)}_{sanitize_filename(fajl.filename)}" with open(uj_fajl_path.lstrip("/"), "wb+") as f: shutil.copyfileobj(fajl.file, f) # ===== minden ellenőrzés rendben: a módosítás ===== valt("név", cel.username, nev) valt("e-mail", cel.email, email) valt("aktív", cel.aktiv, uj_aktiv) valt("csak olvasás", cel.csak_olvasas, uj_csak) valt("hozzáférési szint", cel.hozzaferes_szint or "CEG", szint) valt("telephelyek", ",".join(str(i) for i in sorted(s.id for s in cel.allowed_sites)), ",".join(str(i) for i in sorted(s.id for s in uj_telephelyek))) valt("gépek", ",".join(str(i) for i in sorted(m.id for m in cel.allowed_machines)), ",".join(str(i) for i in sorted(m.id for m in uj_gepek))) for kulcs, cimke in (("jog_uzemora_rogzites", "jog: üzemóra"), ("jog_dokumentum_feltoltes", "jog: dokumentum"), ("jog_gep_szerkesztes", "jog: gép szerkesztés"), ("jog_datum_szerkesztes", "jog: dátum szerkesztés")): valt(cimke, getattr(cel, kulcs), form.get(kulcs) == "on") for kulcs, cimke in (("bizonyitvany_szam", "bizonyítvány szám"), ("kamarai_azonosito", "kamarai azonosító"), ("engedely_szam", "engedély szám"), ("tanusitvany_szam", "tanúsítvány szám")): valt(cimke, getattr(cel, kulcs), (form.get(kulcs) or "").strip()) valt("orvosi alkalmasság érvényesség", cel.orvosi_alkalmassag_ervenyesseg, orvosi) valt("jogosítvány érvényesség", cel.jogositvany_ervenyesseg, jogositvany) valt("modulok", ",".join(sorted(m.kod for m in cel.sajat_modulok)), ",".join(sorted(m.kod for m in uj_modulok))) if uj_fajl_path: valt("bizonyítvány fájl", cel.bizonyitvany_path, uj_fajl_path) cel.username = nev if email != cel.email: cel.email = email aktivalas_utils.email_valtozas_kezelese(db, cel, current_user, regi_email=regi_email) # váró fióknál új link az új címre cel.aktiv = uj_aktiv cel.csak_olvasas = uj_csak cel.hozzaferes_szint = szint cel.allowed_sites = uj_telephelyek cel.allowed_machines = uj_gepek cel.jog_uzemora_rogzites = form.get("jog_uzemora_rogzites") == "on" cel.jog_dokumentum_feltoltes = form.get("jog_dokumentum_feltoltes") == "on" cel.jog_gep_szerkesztes = form.get("jog_gep_szerkesztes") == "on" cel.jog_datum_szerkesztes = form.get("jog_datum_szerkesztes") == "on" cel.bizonyitvany_szam = (form.get("bizonyitvany_szam") or "").strip() cel.kamarai_azonosito = (form.get("kamarai_azonosito") or "").strip() cel.engedely_szam = (form.get("engedely_szam") or "").strip() cel.tanusitvany_szam = (form.get("tanusitvany_szam") or "").strip() cel.orvosi_alkalmassag_ervenyesseg = orvosi cel.jogositvany_ervenyesseg = jogositvany cel.sajat_modulok = uj_modulok if uj_fajl_path: cel.bizonyitvany_path = uj_fajl_path if not cel.aktiv: aktivalas_utils.tokenek_torlese(db, cel) # inaktivált, aktiválásra váró fiók linkje ne maradjon élő for mezo, regi, uj in valtozasok: _naplo(db, cel.company_id, current_user, f"Felhasználó adata: {mezo} (rendszeradmin)", f"{nev}: {regi}", f"{nev}: {uj}") db.commit() if not valtozasok: return JSONResponse(content={"message": "Nem volt változás.", "valtozasok": []}) return JSONResponse(content={"message": f"Mentve ({len(valtozasok)} változás).", "valtozasok": [v[0] for v in valtozasok]}) # --------------------------------------------------------------------------------------------------------------------------- # JELSZÓ: POST /felhasznalok/{id}/jelszo # --------------------------------------------------------------------------------------------------------------------------- @router.post("/felhasznalok/{user_id}/jelszo") async def jelszo_csere( user_id: int, request: Request, uj_jelszo: str = Form(...), uj_jelszo2: str = Form(""), mfa_torles: str = Form(""), sajat_jelszo: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user), ): """Új jelszó beállítása BÁRMELYIK felhasználónak (másik rendszeradminnak is). Opcionálisan (alapból bepipálva) a 2FA-beállítás törlése is (elveszett eszköz). Aktiválásra váró fióknál ez KÉZI AKTIVÁLÁS (a jelszóval a fiók azonnal aktív). Másik RA fiókján: saját jelszó kell. A jelszócsere minden meglévő munkamenetét érvényteleníti (jelszo_valtozott_at).""" cel, hiba = _kapu(request, current_user, db, user_id) if hiba: return hiba if cel.id == current_user.id: return _hiba(403, 'A saját jelszavadat a "Jelszavam módosítása" menüben változtathatod (ott a régi jelszó is kell).') if uj_jelszo != uj_jelszo2: # v27: az új jelszót meg kell erősíteni (elgépelés ellen) return _hiba(400, "A két új jelszó nem egyezik!") if not jelszo_eleg_eros_e(uj_jelszo): return _hiba(400, "A jelszó túl rövid (legalább 6 karakter) vagy túl hosszú.") hazirend_hiba = jelszo_hazirend.ellenorzes(db, cel, uj_jelszo) # v36: a cél cégének jelszó-házirendje (bonyolultság, előzmény) if hazirend_hiba: return _hiba(400, hazirend_hiba) if _ujraazonositas_kell(cel) and not _ujraazonositas_rendben(current_user, sajat_jelszo): return _hiba(403, "Másik rendszeradminisztrátor jelszavának módosításához add meg a SAJÁT jelszavadat (újra-azonosítás).") mfa = mfa_torles in ("on", "1", "true") volt_mfa = mfa_utils.beallitas_van(cel) cel_id, actor_id = cel.id, current_user.id if not aktivalas_utils.aktivalva_e(cel): aktivalas_utils.kezi_aktivalas(db, cel, current_user, uj_jelszo) # COMMITOL, beállítja a jelszót és az aktivált állapotot _naplo(db, cel.company_id, current_user, "Jelszó (rendszeradmin)", f"{cel.username}: aktiválásra várt", f"{cel.username}: kézi aktiválás új jelszóval") db.commit() return JSONResponse(content={"message": "A fiók kézzel aktiválva, a jelszó beállítva."}) regi_hash = cel.hashed_password cel.hashed_password = get_password_hash(uj_jelszo) cel.jelszo_valtozott_at = datetime.now(timezone.utc) jelszo_hazirend.uj_jelszo_utan(db, cel, regi_hash) # v36: előzmény + új lejárat mfa_torolve = False if mfa and volt_mfa: mfa_utils.torles(cel) mfa_torolve = True _naplo(db, cel.company_id, current_user, "Jelszó (rendszeradmin)", f"{cel.username}: -", f"{cel.username}: jelszó megváltoztatva" + ("; 2FA-beállítás törölve" if mfa_torolve else "")) db.commit() mfa_utils.ertesito_visszaallitasrol(db, cel_id, actor_id, jelszo_is=True, mfa_is=mfa_torolve) return JSONResponse(content={"message": "A jelszó megváltoztatva" + (", a 2FA-beállítás törölve." if mfa_torolve else ".")}) # --------------------------------------------------------------------------------------------------------------------------- # 2FA: POST /felhasznalok/{id}/mfa-torles # --------------------------------------------------------------------------------------------------------------------------- @router.post("/felhasznalok/{user_id}/mfa-torles") async def mfa_torles_vegpont( user_id: int, request: Request, sajat_jelszo: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user), ): """A felhasználó 2FA-beállításának törlése/visszaállítása (elveszett eszköz, hibás beállítás, nem olvasható titok). A felhasználó minden munkamenete érvénytelen lesz, és a következő belépéskor (ha a cégnél kötelező) új QR-beállítás kell. Másik RA fiókján: saját jelszó kell. A saját 2FA-d a /rendszeradmin/ketlepcsos oldalon kezelhető (itt tiltott, mert kizárhatnád magad).""" cel, hiba = _kapu(request, current_user, db, user_id) if hiba: return hiba if cel.id == current_user.id: return _hiba(403, "A saját 2FA-beállításodat nem törölheted innen (kizárnád magad). Használd a /rendszeradmin/ketlepcsos oldalt.") if not mfa_utils.beallitas_van(cel): return JSONResponse(content={"message": "A felhasználónak nincs 2FA-beállítása, nincs mit törölni."}) if _ujraazonositas_kell(cel) and not _ujraazonositas_rendben(current_user, sajat_jelszo): return _hiba(403, "Másik rendszeradminisztrátor 2FA-jának törléséhez add meg a SAJÁT jelszavadat (újra-azonosítás).") volt = "megerősített beállítás" if cel.totp_megerositve else "megkezdett (meg nem erősített) beállítás" cel_id, actor_id = cel.id, current_user.id mfa_utils.felhasznalo_visszaallitasa(cel) # törli a 2FA-t ÉS érvényteleníti a munkameneteit _naplo(db, cel.company_id, current_user, "Kétlépcsős hitelesítés", f"{volt} ({cel.email})", "törölve rendszeradmin által (új QR-beállítás kell)") db.commit() mfa_utils.ertesito_visszaallitasrol(db, cel_id, actor_id, jelszo_is=False, mfa_is=True) return JSONResponse(content={"message": "A 2FA-beállítás törölve, a felhasználó munkamenetei érvénytelenek. Következő belépéskor újra be kell állítania."}) # --------------------------------------------------------------------------------------------------------------------------- # AKTÍV / INAKTÍV: POST /felhasznalok/{id}/aktiv # --------------------------------------------------------------------------------------------------------------------------- @router.post("/felhasznalok/{user_id}/aktiv") async def aktiv_beallitas(user_id: int, request: Request, ertek: str = Form(...), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user)): """Felhasználó inaktiválása / reaktiválása (explicit érték: "on" = aktív, "off" = inaktív; a belépés inaktiváláskor azonnal megszűnik).""" cel, hiba = _kapu(request, current_user, db, user_id) if hiba: return hiba if ertek not in ("on", "off"): return _hiba(400, "Érvénytelen érték.") uj = ertek == "on" if cel.id == current_user.id and not uj: return _hiba(403, "A saját fiókodat nem inaktiválhatod.") if is_rendszeradminisztrator(cel) and not uj and not _marad_masik_aktiv_ra(db, cel): return _hiba(403, "Ez az utolsó aktív rendszeradminisztrátor, nem inaktiválható.") if uj and not cel.aktiv and cel.role == "VIVU": masik = db.query(models.User).filter(models.User.company_id == cel.company_id, models.User.role == "VIVU", models.User.aktiv.is_(True), models.User.id != cel.id).first() if masik: return _hiba(400, f'Ennél a cégnél már van aktív VIVÜ ügyintéző ("{masik.username}") - cégenként csak egy lehet.') if uj and not cel.aktiv: licenc_hiba = licenc_reaktivalas_hiba(cel, db) # J-06 (v66): a licenc-limit visszakapcsoláskor is él if licenc_hiba: return _hiba(400, licenc_hiba) if bool(cel.aktiv) == uj: return JSONResponse(content={"message": "Nem volt változás."}) regi = "Aktív" if cel.aktiv else "Inaktív" cel.aktiv = uj _naplo(db, cel.company_id, current_user, "Felhasználó állapota", f"{cel.username}: {regi}", f"{cel.username}: {'Aktív' if uj else 'Inaktív'}") if not uj: aktivalas_utils.tokenek_torlese(db, cel) db.commit() return JSONResponse(content={"message": "A felhasználó " + ("aktiválva." if uj else "inaktiválva.")}) # --------------------------------------------------------------------------------------------------------------------------- # VÉGLEGES TÖRLÉS: POST /felhasznalok/{id}/torles # --------------------------------------------------------------------------------------------------------------------------- @router.post("/felhasznalok/{user_id}/torles") async def vegleges_torles( user_id: int, request: Request, megerositett_email: str = Form(...), megertettem: str = Form(""), sajat_jelszo: str = Form(""), db: Session = Depends(get_db), current_user: models.User = Depends(get_current_user), ): """VÉGLEGES törlés (nincs visszaút). Kétszeres megerősítés (az e-mail pontos begépelése + tudomásulvétel). A rá hivatkozó rekordok (vizsgálatok, naplók stb.) megmaradnak, csak a felhasználóra mutató hivatkozás NULL lesz (rendszeradmin._torli_felhasznalot). MÁSIK RENDSZERADMIN is törölhető (az eddigi tiltás megszűnt), de saját jelszó kell, és az utolsó aktív RA / a saját fiók nem törölhető.""" cel, hiba = _kapu(request, current_user, db, user_id) if hiba: return hiba if cel.id == current_user.id: return _hiba(403, "A saját fiókodat nem törölheted.") if is_rendszeradminisztrator(cel) and not _marad_masik_aktiv_ra(db, cel): return _hiba(403, "Ez az utolsó aktív rendszeradminisztrátor, nem törölhető.") if megertettem != "on": return _hiba(400, "A tudomásulvételi jelölőnégyzet bejelölése kötelező.") if (megerositett_email or "").strip() != (cel.email or ""): return _hiba(400, "A begépelt e-mail cím nem egyezik pontosan - a törlés megszakítva, semmi nem változott.") if _ujraazonositas_kell(cel) and not _ujraazonositas_rendben(current_user, sajat_jelszo): return _hiba(403, "Rendszeradminisztrátor törléséhez add meg a SAJÁT jelszavadat (újra-azonosítás).") from routers.rendszeradmin import _torli_felhasznalot leiras = f"{cel.username} ({cel.email}), szerepkör: {cel.role}, cég: #{cel.company_id}" ceg_id, email = cel.company_id, cel.email ra_volt = is_rendszeradminisztrator(cel) try: _torli_felhasznalot(cel.id, db) _naplo(db, ceg_id, current_user, "Felhasználó törölve (rendszeradmin)", leiras, "véglegesen törölve") db.commit() except Exception as ex: db.rollback() print(f"❌ Hiba a felhasználó törlése közben: {type(ex).__name__}: {ex}") return _hiba(500, "Hiba történt a törlés közben, semmi nem került mentésre.") print(f"[VÉGLEGES FELHASZNÁLÓ-TÖRLÉS] '{email}' fiókot véglegesen törölte: {current_user.username} ({current_user.email}){' - RENDSZERADMIN fiók' if ra_volt else ''}") return JSONResponse(content={"message": f"'{email}' véglegesen törölve."})