"""Authentication service - password hashing, API key management.""" import hashlib import secrets from datetime import datetime, timedelta from src.backend.utils.tempo import utcnow import bcrypt from fastapi import HTTPException, status from sqlalchemy import delete, select, update from sqlalchemy.ext.asyncio import AsyncSession from src.backend.config import settings from src.backend.models.orm.api_key import ApiKey from src.backend.models.orm.user import User def hash_password(password: str) -> str: """Hash a password using bcrypt.""" return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8") def verify_password(plain_password: str, hashed_password: str) -> bool: """Verify a password against its hash.""" return bcrypt.checkpw( plain_password.encode("utf-8"), hashed_password.encode("utf-8") ) def generate_api_key() -> str: """Generate a secure 64-character API key.""" return secrets.token_urlsafe(48) # 64 chars base64 # A precomputed, unused bcrypt hash. When the username does not match any # user, authenticate_user still runs bcrypt against this instead of # returning immediately, so "unknown username" costs the same as "wrong # password" and a caller cannot use response time to enumerate usernames. _DUMMY_PASSWORD_HASH = "$2b$12$BLtwrR2COyd5W/UWfxp7BOc7OHua9jazo6zZdWOwRD9PV.pyaqKZC" async def authenticate_user( db: AsyncSession, username: str, password: str ) -> User | None: """Authenticate user by username and password.""" result = await db.execute( select(User).where(User.username == username, User.active == True) ) user = result.scalar_one_or_none() if user is None: verify_password(password, _DUMMY_PASSWORD_HASH) return None if not verify_password(password, user.password_hash): return None return user # #59: PIN a 4 cifre. Hash identico alle password (bcrypt); 5 errori -> blocco 15 minuti. PIN_MAX_FAILURES = 5 PIN_LOCK_MINUTES = 15 async def set_user_pin(db: AsyncSession, user: User, pin: str | None) -> None: """Imposta o toglie (None) il PIN; azzera contatore e blocco. Admin esclusi (#59).""" if pin is not None and user.is_admin: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Admin users cannot have a PIN", ) user.pin_hash = hash_password(pin) if pin is not None else None user.pin_failed_count = 0 user.pin_locked_until = None await db.flush() async def authenticate_pin(db: AsyncSession, user_id: int, pin: str) -> User: """Verifica user_id+PIN con blocco dopo 5 errori consecutivi (#59). 401 PIN errato (o utente non idoneo: stessa risposta, niente enumerazione), 423 utente bloccato. Il contatore va committato prima di sollevare: get_db fa rollback su eccezione e l'errore andrebbe perso. """ result = await db.execute( select(User).where(User.id == user_id, User.active == True) # noqa: E712 ) user = result.scalar_one_or_none() if user is None or user.is_admin or user.pin_hash is None: verify_password(pin, _DUMMY_PASSWORD_HASH) # stesso costo di un PIN errato raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid PIN" ) now = utcnow() if user.pin_locked_until is not None: if user.pin_locked_until > now: raise HTTPException( status_code=status.HTTP_423_LOCKED, detail="PIN locked until " + user.pin_locked_until.strftime("%Y-%m-%d %H:%M:%S") + " UTC", ) # blocco scaduto: si riparte da zero user.pin_locked_until = None user.pin_failed_count = 0 if verify_password(pin, user.pin_hash): if user.pin_failed_count: user.pin_failed_count = 0 await db.flush() return user user.pin_failed_count += 1 locked = user.pin_failed_count >= PIN_MAX_FAILURES if locked: user.pin_locked_until = now + timedelta(minutes=PIN_LOCK_MINUTES) await db.commit() if locked: raise HTTPException( status_code=status.HTTP_423_LOCKED, detail="PIN locked until " + user.pin_locked_until.strftime("%Y-%m-%d %H:%M:%S") + " UTC", ) raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid PIN") async def authorise_supervisor( db: AsyncSession, username: str | None = None, password: str | None = None, user_id: int | None = None, pin: str | None = None, ) -> User: """Check the credentials of the supervisor authorising something. Used wherever the shop floor needs a capoturno: a line stop, a closure, an out-of-tolerance value allowed to stand. Lives here rather than beside any one of them because it is the same question every time - who is this, and are they allowed to say yes. Uses authenticate_user rather than a full login: logging in would issue a needless API key for the supervisor. """ if user_id is not None and pin is not None: # #59: via PIN, stesso blocco del login con PIN (401/423 sollevati da li') supervisor = await authenticate_pin(db, user_id, pin) elif username is not None and password is not None: supervisor = await authenticate_user(db, username, password) else: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="Supervisor credentials required", ) if supervisor is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid supervisor credentials", ) if not supervisor.has_role("Supervisor") and not supervisor.is_admin: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="User is not authorised as supervisor", ) return supervisor def hash_api_key(key: str) -> str: """SHA-256 hex digest of an API key; this is all the DB ever stores.""" return hashlib.sha256(key.encode("utf-8")).hexdigest() async def issue_api_key(db: AsyncSession, user: User) -> str: """Create a new session key for the user and return it in clear (once).""" key = generate_api_key() now = utcnow() # Housekeeping: drop this user's expired keys so the table does not grow forever. await db.execute( delete(ApiKey).where(ApiKey.user_id == user.id, ApiKey.expires_at <= now) ) db.add( ApiKey( user_id=user.id, key_hash=hash_api_key(key), created_at=now, expires_at=now + timedelta(hours=settings.api_key_ttl_hours), last_used_at=now, ) ) await db.flush() return key async def revoke_all_keys(db: AsyncSession, user_id: int) -> None: """Invalidate every session key of a user.""" await db.execute(delete(ApiKey).where(ApiKey.user_id == user_id)) await db.flush() async def login_user(db: AsyncSession, user: User) -> str: """Issue a new API key for this login and update last_login. One key per login: the same account used from several stations at once (#25) keeps every station working, because no login revokes another's key. """ api_key = await issue_api_key(db, user) await db.execute( update(User).where(User.id == user.id).values(last_login=utcnow()) ) await db.flush() return api_key async def logout_user( db: AsyncSession, user: User, key: str | None = None, all_sessions: bool = False ) -> None: """Log out. Deletes only the key of this session; all_sessions=True deletes every key of the user.""" if all_sessions: await revoke_all_keys(db, user.id) return if key: await db.execute( delete(ApiKey).where( ApiKey.user_id == user.id, ApiKey.key_hash == hash_api_key(key) ) ) await db.flush() async def create_user( db: AsyncSession, username: str, password: str, display_name: str, email: str | None = None, roles: list[str] | None = None, is_admin: bool = False, language_pref: str = "it", theme_pref: str = "light", ) -> User: """Create a new user with hashed password.""" user = User( username=username, password_hash=hash_password(password), display_name=display_name, email=email, roles=roles or [], is_admin=is_admin, language_pref=language_pref, theme_pref=theme_pref, ) db.add(user) await db.flush() await db.refresh(user) return user async def regenerate_api_key(db: AsyncSession, user: User) -> str: """Revoke every key of the user and issue a fresh one.""" await revoke_all_keys(db, user.id) return await issue_api_key(db, user)