leocrm/app/services/auth_service.py

"""Authentication service — login, logout, password reset, session management."""

from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime, timedelta
from typing import Any

import redis.asyncio as aioredis
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import get_settings
from app.core.audit import log_audit
from app.core.auth import (
    create_session,
    get_redis,
    get_session_data,
    hash_password,
    hash_token,
    invalidate_session,
    update_session_tenant,
    verify_password,
)
from app.core.hooks import do_action, apply_filters
from app.models.auth import PasswordResetToken
from app.models.tenant import Tenant
from app.models.user import User, UserTenant

logger = logging.getLogger(__name__)


class AuthService:
    """Handles authentication operations."""

    async def login(
        self,
        db: AsyncSession,
        redis: aioredis.Redis,
        email: str,
        password: str,
        tenant_slug: str | None = None,
    ) -> tuple[str, str, User, Tenant, str] | None:
        """Authenticate user and create session.
        Returns (session_id, csrf_token, user, tenant, role) or None.

        Email is globally unique so we can safely use scalar_one_or_none().
        The tenant is resolved from UserTenant via tenant_slug or the
        user's default tenant membership.
        """
        # Hook: auth.before_login — filter can modify email
        email = await apply_filters("auth.before_login", email, db=db, password=password, tenant_slug=tenant_slug)

        # Find user by email (globally unique now)
        q = select(User).where(User.email == email, User.is_active == True)  # noqa: E712
        result = await db.execute(q)
        user = result.scalar_one_or_none()
        if user is None:
            return None

        if not verify_password(password, user.password_hash):
            return None

        # Get user's default tenant or the one matching slug
        ut_q = select(UserTenant).where(UserTenant.user_id == user.id)
        if tenant_slug:
            ut_q = ut_q.join(Tenant, UserTenant.tenant_id == Tenant.id).where(
                Tenant.slug == tenant_slug
            )
        else:
            ut_q = ut_q.where(UserTenant.is_default == True)  # noqa: E712
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()

        # Fallback: just get first tenant membership
        if user_tenant is None:
            ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
            ut_result2 = await db.execute(ut_q2)
            user_tenant = ut_result2.scalar_one_or_none()
            if user_tenant is None:
                return None

        tenant_q = select(Tenant).where(Tenant.id == user_tenant.tenant_id)
        tenant_result = await db.execute(tenant_q)
        tenant = tenant_result.scalar_one_or_none()
        if tenant is None:
            return None

        session_id, csrf_token = await create_session(
            db, redis, user, tenant.id, role=user_tenant.role
        )

        # Log the login in audit trail
        await log_audit(
            db,
            tenant.id,
            user.id,
            "login",
            "user",
            user.id,
            changes={"email": email},
        )

        # Hook: auth.after_login
        await do_action("auth.after_login", db=db, user=user, tenant=tenant, role=user_tenant.role, session_id=session_id)

        return session_id, csrf_token, user, tenant, user_tenant.role

    async def logout(self, redis: aioredis.Redis, session_id: str) -> bool:
        """Invalidate a session."""
        await invalidate_session(redis, session_id)
        return True

    async def get_current_user_info(
        self,
        db: AsyncSession,
        redis: aioredis.Redis,
        session_id: str,
    ) -> dict[str, Any] | None:
        """Get current user info from session."""
        session_data = await get_session_data(redis, session_id)
        if session_data is None:
            return None

        # Fetch tenant name
        tenant_q = select(Tenant).where(Tenant.id == uuid.UUID(session_data["tenant_id"]))
        tenant_result = await db.execute(tenant_q)
        tenant = tenant_result.scalar_one_or_none()

        return {
            "user_id": session_data["user_id"],
            "email": session_data["email"],
            "name": session_data["name"],
            "role": session_data["role"],
            "is_system_admin": session_data.get("is_system_admin", False),
            "tenant_id": session_data["tenant_id"],
            "tenant_name": tenant.name if tenant else None,
        }

    async def switch_tenant(
        self,
        db: AsyncSession,
        redis: aioredis.Redis,
        session_id: str,
        new_tenant_id: uuid.UUID,
    ) -> dict[str, Any] | None:
        """Switch the active tenant for the current session."""
        session_data = await get_session_data(redis, session_id)
        if session_data is None:
            return None

        user_id = uuid.UUID(session_data["user_id"])

        # Verify user is member of target tenant
        ut_q = select(UserTenant).where(
            UserTenant.user_id == user_id,
            UserTenant.tenant_id == new_tenant_id,
        )
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()
        if user_tenant is None:
            return None

        updated = await update_session_tenant(redis, session_id, new_tenant_id, role=user_tenant.role)
        if updated is None:
            return None

        # Fetch tenant name
        tenant_q = select(Tenant).where(Tenant.id == new_tenant_id)
        tenant_result = await db.execute(tenant_q)
        tenant = tenant_result.scalar_one_or_none()

        updated["tenant_name"] = tenant.name if tenant else None
        return updated

    async def request_password_reset(
        self,
        db: AsyncSession,
        email: str,
        tenant_id: uuid.UUID | None = None,
    ) -> bool:
        """Create a password reset token. Always returns True (no user enumeration)."""
        q = select(User).where(User.email == email, User.is_active == True)  # noqa: E712
        result = await db.execute(q)
        user = result.scalar_one_or_none()
        if user is None:
            return True  # Don't reveal whether email exists

        # Resolve tenant_id from UserTenant (default or specified)
        ut_q = select(UserTenant).where(UserTenant.user_id == user.id)
        if tenant_id is not None:
            ut_q = ut_q.where(UserTenant.tenant_id == tenant_id)
        else:
            ut_q = ut_q.where(UserTenant.is_default == True)  # noqa: E712
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()
        if user_tenant is None:
            # Fallback: get first tenant membership
            ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
            ut_result2 = await db.execute(ut_q2)
            user_tenant = ut_result2.scalar_one_or_none()
            if user_tenant is None:
                return True

        # Invalidate previous unused tokens
        prev_q = select(PasswordResetToken).where(
            PasswordResetToken.user_id == user.id,
            PasswordResetToken.used_at.is_(None),
        )
        prev_result = await db.execute(prev_q)
        for prev_token in prev_result.scalars().all():
            prev_token.used_at = datetime.now(UTC)

        # Create new token
        import secrets

        raw_token = secrets.token_urlsafe(32)
        token_hash = hash_token(raw_token)
        settings = get_settings()
        expires_at = datetime.now(UTC) + timedelta(hours=settings.password_reset_expiry_hours)

        reset_token = PasswordResetToken(
            tenant_id=user_tenant.tenant_id,
            user_id=user.id,
            token_hash=token_hash,
            expires_at=expires_at,
        )
        db.add(reset_token)
        await db.flush()

        # Enqueue ARQ job to send the password reset email
        try:
            from app.core.jobs import enqueue_job

            await enqueue_job(
                "send_password_reset_email",
                user_id=str(user.id),
                email=user.email,
                raw_token=raw_token,
                expires_at=expires_at.isoformat(),
            )
            logger.info("Enqueued password reset email job for user %s", user.id)
        except Exception:
            logger.warning(
                "ARQ enqueue failed for password reset email — "
                "email will not be sent. Check Redis/ARQ connectivity.",
                exc_info=True,
            )

        return True

    async def confirm_password_reset(
        self,
        db: AsyncSession,
        token: str,
        new_password: str,
    ) -> bool:
        """Reset password using a valid token. Returns True on success."""
        token_hash = hash_token(token)
        q = select(PasswordResetToken).where(
            PasswordResetToken.token_hash == token_hash,
            PasswordResetToken.used_at.is_(None),
        )
        result = await db.execute(q)
        reset_token = result.scalar_one_or_none()

        if reset_token is None:
            return False

        if reset_token.expires_at < datetime.now(UTC):
            return False  # Token expired

        # Get user
        user_q = select(User).where(User.id == reset_token.user_id)
        user_result = await db.execute(user_q)
        user = user_result.scalar_one_or_none()
        if user is None:
            return False

        # Update password
        user.password_hash = hash_password(new_password)
        reset_token.used_at = datetime.now(UTC)
        await db.flush()

        # Invalidate all active Redis sessions for this user
        try:
            redis = get_redis()
            # Scan for session keys and check which belong to this user
            import json

            async for key in redis.scan_iter(match="session:*", count=100):
                raw = await redis.get(key)
                if raw is None:
                    continue
                try:
                    session_data = json.loads(raw)
                except (json.JSONDecodeError, TypeError):
                    continue
                if session_data.get("user_id") == str(user.id):
                    await redis.delete(key)
                    logger.info("Deleted session %s for user %s after password reset", key, user.id)
        except Exception:
            logger.warning("Failed to invalidate Redis sessions for user %s", user.id, exc_info=True)

        # Audit log entry for password reset
        try:
            await log_audit(
                db,
                reset_token.tenant_id,
                user.id,
                "password_reset",
                "user",
                user.id,
                changes={"action": "password_changed"},
            )
        except Exception:
            logger.warning("Failed to create audit log for password reset of user %s", user.id, exc_info=True)

        return True

    async def get_password_reset_token_raw(self, db: AsyncSession, email: str) -> str | None:
        """Get the raw (unhashed) reset token for testing purposes.
        This simulates what would be sent via email.
        """
        import secrets

        q = select(User).where(User.email == email)
        result = await db.execute(q)
        user = result.scalar_one_or_none()
        if user is None:
            return None

        # Get tenant_id from UserTenant
        ut_q = select(UserTenant).where(
            UserTenant.user_id == user.id,
            UserTenant.is_default == True,  # noqa: E712
        )
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()
        if user_tenant is None:
            ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
            ut_result2 = await db.execute(ut_q2)
            user_tenant = ut_result2.scalar_one_or_none()
            if user_tenant is None:
                return None

        raw_token = secrets.token_urlsafe(32)
        token_hash = hash_token(raw_token)
        settings = get_settings()
        expires_at = datetime.now(UTC) + timedelta(hours=settings.password_reset_expiry_hours)

        reset_token = PasswordResetToken(
            tenant_id=user_tenant.tenant_id,
            user_id=user.id,
            token_hash=token_hash,
            expires_at=expires_at,
        )
        db.add(reset_token)
        await db.flush()
        return raw_token

    async def create_expired_reset_token(self, db: AsyncSession, email: str) -> str | None:
        """Create an already-expired reset token for testing."""
        import secrets

        q = select(User).where(User.email == email)
        result = await db.execute(q)
        user = result.scalar_one_or_none()
        if user is None:
            return None

        # Get tenant_id from UserTenant
        ut_q = select(UserTenant).where(
            UserTenant.user_id == user.id,
            UserTenant.is_default == True,  # noqa: E712
        )
        ut_result = await db.execute(ut_q)
        user_tenant = ut_result.scalar_one_or_none()
        if user_tenant is None:
            ut_q2 = select(UserTenant).where(UserTenant.user_id == user.id)
            ut_result2 = await db.execute(ut_q2)
            user_tenant = ut_result2.scalar_one_or_none()
            if user_tenant is None:
                return None

        raw_token = secrets.token_urlsafe(32)
        token_hash = hash_token(raw_token)
        expires_at = datetime.now(UTC) - timedelta(hours=1)  # Already expired

        reset_token = PasswordResetToken(
            tenant_id=user_tenant.tenant_id,
            user_id=user.id,
            token_hash=token_hash,
            expires_at=expires_at,
        )
        db.add(reset_token)
        await db.flush()
        return raw_token


auth_service = AuthService()