fai'XdZddlZddlZddlmZmZddlmZmZmZGddZ dS)u app/services/auth_service.py ============================ Servizio autenticazione con JWT token. Responsabilità: - Recuperare user da database - Validare password - Generare JWT token - Decodificare e validare JWT - Gestire refresh token N)datetime timedelta)DictOptionalTuplec 0eZdZdZ ddedededefdZd Zdd ed ed ede e ee ee effdZ defdZ dede e ee effdZ dede e ee effdZ dd ed ed ede eeffdZdS) AuthServiceu\ Servizio centralizzato per autenticazione JWT. Parametri iniziali: db_config (dict): Credenziali MySQL {host, user, password, database} jwt_secret (str): Secret key per firmare JWT jwt_algorithm (str): Algoritmo JWT (default: HS256) jwt_expiration_hours (int): Ore di validità token (default: 24) HS256 db_config jwt_secret jwt_algorithmjwt_expiration_hoursc>||_||_||_||_dS)N)r r rr)selfr r rrs F/var/www/enigma.pooltech.it/enigma_inventario/services_auth_service.py__init__zAuthService.__init__s&"$*$8!!!c tj|jd|jd|jd|jddtjj}|S#tj$r$}t dt|d}~wwxYw) z Crea connessione MySQL. Returns: pymysql.Connection: Connessione al database Raises: Exception: Se connessione fallisce hostuserpassworddatabaseutf8mb4)rrrrcharset cursorclasszDatabase connection error: N)pymysqlconnectr cursors DictCursorError Exceptionstr)rconnes r_get_db_connectionzAuthService._get_db_connection&s D?^F+^F+ 3 3!#O6 DK} D D DB#a&&BBCC C DsAAB (BB emailr tenant_idreturncd} |}|5}d}||||f|}dddn #1swxYwY|s"ddd|d|f|r|SSddlm}||} | |s |r|dSdS| | } |5}d}||tj | j f| dddn #1swxYwY| | df|r|SS#t$r6} ddd t!| fcYd} ~ |r|SSd} ~ wwxYw#|r|wwxYw) uU Login user e genera JWT token. Per la DEMO: tenant_id=1 è sempre il tenant di test. Args: email (str): Email utente password (str): Password in chiaro tenant_id (int): ID tenant (default: 1 per demo) Returns: Tuple[str, dict, str]: (access_token, user_dict, error_message) - success: (token, user_data, None) - failure: (None, None, error_message) Esempio: >>> service = AuthService(db_config, jwt_secret='secret123') >>> token, user, error = service.login('operatore@studium.it', 'password123') >>> if token: ... print(f"Login OK: {user['nome']}") ... else: ... print(f"Login failed: {error}") Nz SELECT id, id_tenant, email, nome, password_hash, ruolo FROM tenant_users WHERE email = %s AND id_tenant = %s LIMIT 1 User z not found in tenant rUser)NNzInvalid passwordz9UPDATE tenant_users SET ultimo_accesso = %s WHERE id = %sz Login error: )r&cursorexecutefetchoneclose models_userr. from_dictverify_passwordgenerate_tokenrutcnowidcommitto_dictr"r#) rr(rr)r$r/sqluser_rowr.rtokenr%s rloginzAuthService.login=s0) **,,D -& sUI$6777!??,, - - - - - - - - - - - - - - - ST#R5#R#Ry#R#RR.   ) ) ( ( ( ( (>>(++D''11 65     ''--E &QsX_%6%6$@AAA                 $,,..$.     8 8 87s1vv777 7 7 7 7 7    8   s(F/A' F'A++F.A+/F0F!)F AE FEF"E#F GG2G3GGGG0c|}tjt|jz|d<tj|d<d|d<t j||j|j}|S)at Genera JWT token per user. Args: user: Istanza User (da models.user) Returns: str: JWT token firmato Nota: Il token contiene: - user_id, email, tenant_id, ruolo (dati critici per autorizzazione) - exp: timestamp scadenza - iat: timestamp creazione hoursexpiatBearer token_type algorithm) to_jwt_payloadrr7rrjwtencoder r)rrpayloadr=s rr6zAuthService.generate_tokensx%%''"**YT=V-W-W-WW!** (  7DOt?QRRR rr=c tj||j|jg}|dfS#tj$rYdStj$r}ddt |fcYd}~Sd}~wt$r}ddt |fcYd}~Sd}~wwxYw)aP Valida JWT token e restituisce payload. Args: token (str): JWT token (senza "Bearer " prefix) Returns: Tuple[dict, str]: (payload, error_message) - valid: (payload_dict, None) - invalid: (None, error_message) Esempio: >>> payload, error = service.validate_token(token) >>> if payload: ... print(f"Token valid for user {payload['user_id']}") ... else: ... print(f"Token invalid: {error}") ) algorithmsN)Nz Token expiredzInvalid token: zToken validation error: )rIdecoder rExpiredSignatureErrorInvalidTokenErrorr#r")rr=rKr%s rvalidate_tokenzAuthService.validate_tokens& =jTEWDXYYYGD= ( ) ) )((($ 4 4 433q66333 3 3 3 3 3 3 = = =z-AuthService.refresh_token..s(SSS11N;R;Rq!;R;R;Rrr@rBrCrFzToken refresh error: ) rIrNr rrPr#itemsrr7rrrJr")rrRrKr% new_payload new_tokens r refresh_tokenzAuthService.refresh_tokensO CjDO,0,>+?)5u(=???GG$ C C CB#a&&BBB B B B B B B CTS SSS %_..AZ1[1[1[[ E%_.. E : ;4K]^^^Id? " : : :9Q999 9 9 9 9 9 9 :s8%(AA AA9$C D(D;DDdemo@studium.itdemo123c ddlm}d} |}||}|5}|d||f|}|r"d} || |||fd|d} n#d} || ||d |d fd|d } |d | fcddd|r|SS#1swxYwYnB#t$r5} d dt| fcYd} ~ |r|SSd} ~ wwxYw |r|dSdS#|r|wwxYw)a Crea user di test nel database per la DEMO. SOLO per ambiente di demo! In produzione non usare mai. Args: email (str): Email test user password (str): Password test user tenant_id (int): Tenant ID (default 1) Returns: Tuple[bool, str]: (success, message) Nota: Se user esiste, lo aggiorna. Se non esiste, lo crea. rr-Nz?SELECT id FROM tenant_users WHERE email = %s AND id_tenant = %sz UPDATE tenant_users SET password_hash = %s, data_creazione = NOW() WHERE email = %s AND id_tenant = %s r,z updatedz INSERT INTO tenant_users (id_tenant, email, nome, password_hash, ruolo, data_creazione) VALUES (%s, %s, %s, %s, %s, NOW()) z Demo User operatorez createdTFzError creating test user: ) r3r.r& hash_passwordr/r0r1r9r2r"r#) rr(rr)r.r$pwd_hashr/existingr;messager%s rcreate_test_userzAuthService.create_test_usersH$ %$$$$$, **,,D))(33H" %&UI&"??,,6C NN35)(DEEE5e555GGC NN3!# # )6e555G W}E" %" %" %" %" %" %" %N   O" %" %" %" %" %" %" %" %" %H @ @ @?s1vv??? ? ? ? ? ?    @I" %N    t   sZ=DB C6 D6C::D=C:>DE E D<E E<EEE8N)r r )r')r`rar')__name__ __module__ __qualname____doc__rr#intrr&rrr>r6rQr_boolrhrWrrr r sLN99$9C9 #9EH9999DDD.BB3B#B#BeHUXM[cdh[ikstwkxLxFyBBBBHc6=C=E(4.(3-2O,P====::s:uXc]HSM5Q/R::::@->(1)*AAcA"%A#&A/4T3Y/?AAAAAArr ) rlrIrrrtypingrrrr rWrrrps   ((((((((((((((((((FFFFFFFFFFr