"""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, UserTenant.status == "active", ) 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() # No fallback — if tenant_slug was provided, the membership must exist # and be active in exactly that tenant. If no slug, the default membership # must exist and be active. if user_tenant is None: if tenant_slug: # Specific tenant requested but no active membership — fail return None # No default membership — check if there are multiple active memberships ut_q2 = select(UserTenant).where( UserTenant.user_id == user.id, UserTenant.status == "active", ) ut_result2 = await db.execute(ut_q2) active_memberships = ut_result2.scalars().all() if len(active_memberships) == 1: # Exactly one active membership — use it user_tenant = active_memberships[0] elif len(active_memberships) == 0: return None else: # Multiple active memberships without a default — must specify tenant_slug 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 via separate API session (crm_api with tenant context) # crm_auth must not write to tenant tables — audit_log is a tenant table try: from app.core.db import get_session_factory, set_tenant_context api_factory = get_session_factory() async with api_factory() as audit_db: await set_tenant_context(audit_db, tenant.id) await log_audit( audit_db, tenant.id, user.id, "login", "user", user.id, changes={"email": email}, ) await audit_db.commit() except Exception: logger.warning("Failed to write login audit log via API session", exc_info=True) # 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 has an ACTIVE membership in target tenant ut_q = select(UserTenant).where( UserTenant.user_id == user_id, UserTenant.tenant_id == new_tenant_id, UserTenant.status == "active", ) 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) # Set tenant context for RLS (auth session uses crm_auth role) from app.core.db import set_tenant_context await set_tenant_context(db, user_tenant.tenant_id) # 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 — use separate API session (crm_api) # to avoid requiring audit_log INSERT grants on crm_auth try: from app.core.db import get_api_engine, set_tenant_context from sqlalchemy.ext.asyncio import AsyncSession api_engine = get_api_engine() async with AsyncSession(api_engine) as audit_db: await set_tenant_context(audit_db, reset_token.tenant_id) await log_audit( audit_db, reset_token.tenant_id, user.id, "password_reset", "user", user.id, changes={"action": "password_changed"}, ) await audit_db.commit() 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()