""" 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 """ import jwt import pymysql from datetime import datetime, timedelta from typing import Dict, Optional, Tuple class AuthService: """ 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) """ def __init__(self, db_config: Dict, jwt_secret: str, jwt_algorithm: str = 'HS256', jwt_expiration_hours: int = 24): self.db_config = db_config self.jwt_secret = jwt_secret self.jwt_algorithm = jwt_algorithm self.jwt_expiration_hours = jwt_expiration_hours def _get_db_connection(self): """ Crea connessione MySQL. Returns: pymysql.Connection: Connessione al database Raises: Exception: Se connessione fallisce """ try: conn = pymysql.connect( host=self.db_config['host'], user=self.db_config['user'], password=self.db_config['password'], database=self.db_config['database'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor ) return conn except pymysql.Error as e: raise Exception(f"Database connection error: {str(e)}") def login(self, email: str, password: str, tenant_id: int = 1) -> Tuple[Optional[str], Optional[Dict], Optional[str]]: """ 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}") """ conn = None try: conn = self._get_db_connection() # Query: recupera user dal database # IMPORTANTE: WHERE id_tenant = ? per isolamento multi-tenant with conn.cursor() as cursor: sql = """ SELECT id, id_tenant, email, nome, password_hash, ruolo FROM tenant_users WHERE email = %s AND id_tenant = %s LIMIT 1 """ cursor.execute(sql, (email, tenant_id)) user_row = cursor.fetchone() # User non trovato if not user_row: return None, None, f"User {email} not found in tenant {tenant_id}" # Verifica password from models_user import User user = User.from_dict(user_row) if not user.verify_password(password): return None, None, "Invalid password" # Genera JWT token token = self.generate_token(user) # Update ultimo_accesso nel database with conn.cursor() as cursor: sql = "UPDATE tenant_users SET ultimo_accesso = %s WHERE id = %s" cursor.execute(sql, (datetime.utcnow(), user.id)) conn.commit() return token, user.to_dict(), None except Exception as e: return None, None, f"Login error: {str(e)}" finally: if conn: conn.close() def generate_token(self, user) -> str: """ 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 """ payload = user.to_jwt_payload() # Aggiungi metadata JWT payload['exp'] = datetime.utcnow() + timedelta(hours=self.jwt_expiration_hours) payload['iat'] = datetime.utcnow() payload['token_type'] = 'Bearer' # Firma token token = jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm) return token def validate_token(self, token: str) -> Tuple[Optional[Dict], Optional[str]]: """ 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}") """ try: payload = jwt.decode(token, self.jwt_secret, algorithms=[self.jwt_algorithm]) return payload, None except jwt.ExpiredSignatureError: return None, "Token expired" except jwt.InvalidTokenError as e: return None, f"Invalid token: {str(e)}" except Exception as e: return None, f"Token validation error: {str(e)}" def refresh_token(self, old_token: str) -> Tuple[Optional[str], Optional[str]]: """ Refresh JWT token (genera nuovo token). Args: old_token (str): Token attuale (anche se scaduto, validiamo comunque) Returns: Tuple[str, str]: (new_token, error_message) Nota: Questo permette al client di ottenere un nuovo token senza re-autenticarsi se il vecchio è scaduto. """ # Valida token (anche se scaduto, vogliamo il payload) try: payload = jwt.decode(old_token, self.jwt_secret, algorithms=[self.jwt_algorithm], options={"verify_exp": False}) # Ignora scadenza per ora except jwt.InvalidTokenError as e: return None, f"Cannot refresh invalid token: {str(e)}" # Crea nuovo payload con stessa info ma exp nuovo new_payload = {k: v for k, v in payload.items() if k not in ['exp', 'iat']} new_payload['exp'] = datetime.utcnow() + timedelta(hours=self.jwt_expiration_hours) new_payload['iat'] = datetime.utcnow() try: new_token = jwt.encode(new_payload, self.jwt_secret, algorithm=self.jwt_algorithm) return new_token, None except Exception as e: return None, f"Token refresh error: {str(e)}" def create_test_user(self, email: str = "demo@studium.it", password: str = "demo123", tenant_id: int = 1) -> Tuple[bool, str]: """ 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. """ from models_user import User conn = None try: conn = self._get_db_connection() pwd_hash = User.hash_password(password) with conn.cursor() as cursor: # Controlla se esiste cursor.execute( "SELECT id FROM tenant_users WHERE email = %s AND id_tenant = %s", (email, tenant_id) ) existing = cursor.fetchone() if existing: # Aggiorna sql = """ UPDATE tenant_users SET password_hash = %s, data_creazione = NOW() WHERE email = %s AND id_tenant = %s """ cursor.execute(sql, (pwd_hash, email, tenant_id)) message = f"User {email} updated" else: # Crea nuovo sql = """ INSERT INTO tenant_users (id_tenant, email, nome, password_hash, ruolo, data_creazione) VALUES (%s, %s, %s, %s, %s, NOW()) """ cursor.execute(sql, ( tenant_id, email, "Demo User", pwd_hash, "operatore" )) message = f"User {email} created" conn.commit() return True, message except Exception as e: return False, f"Error creating test user: {str(e)}" finally: if conn: conn.close()