TieMeasureFlow/src/backend/services/auth_service.py

"""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


async def authorise_supervisor(db: AsyncSession, username: str, password: str) -> 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.
    """
    supervisor = await authenticate_user(db, username, password)
    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)