"""Auth domain - business logic orchestration.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Optional, Tuple, Union from urllib.parse import urlencode from uuid import UUID from src.config import get_settings from src.core.exceptions import UnauthorizedError from src.core.protocols import EmailProviderProtocol from src.core.security.hashing import ( digest_token_for_lookup, generate_magic_link_token, generate_refresh_token, generate_6digit_code, hash_token, verify_password_hash, ) from src.core.security.jwt import ( create_access_token, create_refresh_token as create_refresh_jwt, create_mfa_pending_token, decode_token, get_user_id_from_token, ) from src.domain.auth.exceptions import InvalidMagicLinkError, InvalidRefreshTokenError, SignupRequiredError from src.domain.auth.mfa_repository import TotpSecretRepository from src.domain.auth.models import User from src.domain.auth.promotion_service import PromotionService from src.domain.auth.repository import ( MagicLinkRepository, PendingRegistrationRepository, SessionRepository, UserRepository, ) from src.domain.auth.schemas import MfaRequiredResponse, MfaUserInfo, TokenResponse, UserResponse from src.domain.auth.token_store import RedisTokenStore from src.domain.auth.totp_service import TotpService settings = get_settings() # Trusted device cookie name (InTalk-style) TRUSTED_DEVICE_COOKIE = "p2hs_trusted_device" class AuthService: """Auth service - magic link flow, JWT, session management (InTalk-aligned).""" def __init__( self, user_repo: UserRepository, session_repo: SessionRepository, magic_link_repo: MagicLinkRepository, totp_repo: Optional[TotpSecretRepository] = None, email_provider: Optional[EmailProviderProtocol] = None, trusted_device_repo=None, backup_code_service=None, token_store=None, pending_repo: Optional[PendingRegistrationRepository] = None, promotion_service: Optional[PromotionService] = None, ): self._user_repo = user_repo self._session_repo = session_repo self._magic_link_repo = magic_link_repo self._totp_service = TotpService(totp_repo) if totp_repo else None self._email_provider = email_provider self._trusted_device = trusted_device_repo self._backup_codes = backup_code_service self._token_store = token_store self._pending_repo = pending_repo self._promotion_service = promotion_service def _magic_link_url(self, token: str, app_base_url: Optional[str] = None) -> str: """Build full magic link URL for email. Uses app_base_url if provided, else settings.app_url.""" base = (app_base_url or settings.app_url).rstrip("/") return f"{base}/auth/verify?token={token}" def _mobile_magic_link_url(self, token: str) -> str: """Return a fixed first-party app deep link; never accept a caller-provided target.""" separator = "&" if "?" in settings.mobile_magic_link_redirect_uri else "?" return f"{settings.mobile_magic_link_redirect_uri}{separator}{urlencode({'token': token})}" async def request_magic_link( self, email: str, ip_address: Optional[str] = None, app_base_url: Optional[str] = None, *, allow_unknown_email: bool = False, is_signup: bool = False, magic_link_url: Optional[str] = None, token_override: Optional[str] = None, ) -> str: """ Create magic link token. If email provider configured, sends email. When allow_unknown_email=False (login flow): only send if user or pending exists (anti-enumeration). When allow_unknown_email=True (signup flow): always create token and send. is_signup=True sends the welcome/signup email; False sends the returning-user login email. Returns plain token (for caller to send, or for testing). """ from src.domain.auth.email_templates import build_magic_link_email email_lower = email.lower() if not allow_unknown_email and self._pending_repo: user_exists = await self._user_repo.get_by_email(email_lower) is not None pending_exists = await self._pending_repo.get_valid_by_email(email_lower) is not None if not user_exists and not pending_exists: signup_url = f"{settings.home_url.rstrip('/')}/signup?{urlencode({'email': email_lower})}" raise SignupRequiredError(signup_url) token = token_override or generate_magic_link_token() token_digest = digest_token_for_lookup(token) expires_at = datetime.now(timezone.utc) + timedelta(minutes=settings.magic_link_ttl_minutes) await self._magic_link_repo.create( email=email.lower(), token_hash=token_digest, expires_at=expires_at, ip_address=ip_address, ) base_url = app_base_url or settings.app_url if self._email_provider and base_url: link_url = magic_link_url or self._magic_link_url(token, app_base_url=base_url) subject, html, text = build_magic_link_email( link_url, ttl_minutes=settings.magic_link_ttl_minutes, is_signup=is_signup, ) await self._email_provider.send( to=email.lower(), subject=subject, html=html, text=text, ) return token async def request_mobile_magic_link( self, email: str, ip_address: Optional[str] = None, ) -> str: """Send a magic link to the configured P2HS mobile app deep link.""" token = generate_magic_link_token() return await self.request_magic_link( email, ip_address=ip_address, magic_link_url=self._mobile_magic_link_url(token), # Use the generated token so the email link and persisted token match. token_override=token, ) async def password_login( self, email: str, password: str, *, ip_address: Optional[str] = None, user_agent: Optional[str] = None, ) -> Tuple[TokenResponse, User]: """Authenticate an explicitly enabled password-login account. Password login is opt-in per user and is intended for reviewer/service accounts. Normal P2HS users remain magic-link-only. """ user = await self._user_repo.get_by_email(email.lower()) if ( not user or not user.password_login_enabled or not user.password_hash or user.email_verified_at is None or not verify_password_hash(password, user.password_hash) ): raise UnauthorizedError("Invalid email or password") now = datetime.now(timezone.utc) await self._user_repo.update_last_login(user.id, now) refresh_token = generate_refresh_token() refresh_hash = hash_token(refresh_token) expires_at = now + timedelta(days=settings.jwt_refresh_ttl_days) session = await self._session_repo.create( user_id=user.id, refresh_token_hash=refresh_hash, expires_at=expires_at, ip_address=ip_address, user_agent=user_agent, ) access_token = create_access_token(user.id, session.id) refresh_jwt = create_refresh_jwt(user.id, session.id) return TokenResponse( access_token=access_token, refresh_token=refresh_jwt, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ), user async def verify_magic_link( self, token: str, device_info: Optional[dict] = None, ip_address: Optional[str] = None, user_agent: Optional[str] = None, device_id: Optional[str] = None, ) -> Union[Tuple[TokenResponse, User], MfaRequiredResponse]: """Verify magic link token. On success without TOTP: ``(TokenResponse, User)`` so HTTP layer can attach bootstrap. If user has TOTP: ``MfaRequiredResponse``. Trusted device skips TOTP. """ token = (token or "").strip() if not token: raise InvalidMagicLinkError() token_digest = digest_token_for_lookup(token) magic_token = await self._magic_link_repo.get_valid_token(token_digest) if not magic_token: raise InvalidMagicLinkError() email = magic_token.email user = await self._user_repo.get_by_email(email) if not user and self._pending_repo and self._promotion_service: # Deferred signup: promote pending to user pending = await self._pending_repo.get_valid_by_email(email) if pending: user = await self._promotion_service.promote(pending) else: raise InvalidMagicLinkError() elif not user: # Legacy fallback: create user (no pending flow) user = await self._user_repo.create(email) # On first successful magic-link verification: set email_verified_at, update last_login_at now = datetime.now(timezone.utc) if user.email_verified_at is None: await self._user_repo.update_email_verified(user.id) await self._user_repo.update_last_login(user.id, now) await self._magic_link_repo.mark_used(magic_token.id) refresh_token = generate_refresh_token() refresh_hash = hash_token(refresh_token) expires_at = datetime.now(timezone.utc) + timedelta(days=settings.jwt_refresh_ttl_days) session = await self._session_repo.create( user_id=user.id, refresh_token_hash=refresh_hash, expires_at=expires_at, device_info=device_info, ip_address=ip_address, user_agent=user_agent, ) totp_enabled = self._totp_service and await self._totp_service.is_enabled(user.id) if totp_enabled and self._trusted_device and device_id: if await self._trusted_device.is_trusted(user.id, device_id): totp_enabled = False if totp_enabled: mfa_token = create_mfa_pending_token(user.id, session.id) user_info = MfaUserInfo(id=user.id, email=user.email, totp_enabled=True) return MfaRequiredResponse( requires_totp=True, mfa_token=mfa_token, login_token=mfa_token, # InTalk compatibility user=user_info, ) access_token = create_access_token(user.id, session.id) refresh_jwt = create_refresh_jwt(user.id, session.id) now = datetime.now(timezone.utc) tokens = TokenResponse( access_token=access_token, refresh_token=refresh_jwt, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ) return (tokens, user) async def refresh_tokens(self, refresh_token: str) -> TokenResponse: """Refresh access token using refresh token.""" payload = decode_token(refresh_token) if not payload or payload.type != "refresh": raise InvalidRefreshTokenError() try: user_id = UUID(payload.sub) session_id = UUID(payload.session_id) if payload.session_id else None except (ValueError, TypeError): raise InvalidRefreshTokenError() if not session_id: raise InvalidRefreshTokenError() session = await self._session_repo.get_by_id(session_id) if not session or session.revoked_at or session.expires_at < datetime.now(timezone.utc): raise InvalidRefreshTokenError() user = await self._user_repo.get_by_id(user_id) if not user: raise InvalidRefreshTokenError() access_token = create_access_token(user.id, session.id) new_refresh = create_refresh_jwt(user.id, session.id) now = datetime.now(timezone.utc) return TokenResponse( access_token=access_token, refresh_token=new_refresh, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ) async def logout(self, refresh_token: Optional[str] = None, session_id: Optional[UUID] = None) -> None: """Revoke session.""" if session_id: await self._session_repo.revoke(session_id) return if refresh_token: payload = decode_token(refresh_token) if payload and payload.session_id: try: sid = UUID(payload.session_id) await self._session_repo.revoke(sid) except ValueError: pass async def get_current_user(self, access_token: str) -> Optional[User]: """Get user from access token.""" user_id = get_user_id_from_token(access_token) if not user_id: return None return await self._user_repo.get_by_id(user_id) async def complete_totp_mfa( self, mfa_token: str, code: str, remember_device: bool = False, device_id: Optional[str] = None, user_agent: Optional[str] = None, ip_address: Optional[str] = None, ) -> Optional[tuple[TokenResponse, Optional[str], bool, Optional[int]]]: """ Verify TOTP or backup code. Returns (TokenResponse, trusted_device_id, used_backup_code, backup_codes_remaining) or None. """ from src.core.security.jwt import decode_mfa_pending_token decoded = decode_mfa_pending_token(mfa_token) if not decoded: return None user_id, session_id = decoded verified = False used_backup_code = False if self._totp_service: raw = str(code).strip() normalized = "".join(ch for ch in raw.upper() if ch.isalnum()) looks_like_backup = any(ch.isalpha() for ch in normalized) or len(normalized) != 6 if looks_like_backup and self._backup_codes: used_backup_code = await self._backup_codes.consume(user_id, raw) verified = used_backup_code else: verified = await self._totp_service.verify(user_id, code) if not verified: return None session = await self._session_repo.get_by_id(session_id) if not session or session.revoked_at or session.expires_at < datetime.now(timezone.utc): return None user = await self._user_repo.get_by_id(user_id) if not user: return None trusted_device_id = None if remember_device and self._trusted_device: trusted_device_id = await self._trusted_device.trust_device( user_id, device_id=device_id, days=30, user_agent=user_agent, ip_address=ip_address, ) access_token = create_access_token(user.id, session.id) refresh_jwt = create_refresh_jwt(user.id, session.id) now = datetime.now(timezone.utc) tokens = TokenResponse( access_token=access_token, refresh_token=refresh_jwt, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ) remaining = None if used_backup_code and self._backup_codes: remaining = await self._backup_codes.remaining_count(user_id) return (tokens, trusted_device_id, used_backup_code, remaining) async def send_totp_recovery_link(self, email: str, app_base_url: Optional[str] = None) -> dict: """Send TOTP recovery link for lockout. Returns {sent: bool, reason?: str}.""" user = await self._user_repo.get_by_email(email.lower()) if not user: return {"sent": False, "reason": "account_not_found"} if not await self._totp_service.is_enabled(user.id): return {"sent": False, "reason": "totp_not_enabled"} if not self._token_store: return {"sent": False, "reason": "recovery_unavailable"} from src.core.security.hashing import generate_secure_token token = generate_secure_token(48) await self._token_store.store_token( RedisTokenStore.PREFIX_TOTP_RECOVERY, token, {"user_id": str(user.id)}, 900, ) base_url = app_base_url or settings.app_url if self._email_provider and base_url: from src.domain.auth.email_templates import build_totp_recovery_link_email link = f"{base_url.rstrip('/')}/auth/totp-recovery?token={token}" subject, html, text = build_totp_recovery_link_email(link) await self._email_provider.send( to=email.lower(), subject=subject, html=html, text=text, ) return {"sent": True} async def verify_totp_recovery(self, token: str) -> Optional[tuple[User, TokenResponse]]: """Consume recovery token, disable TOTP, create session, return (user, tokens) or None.""" if not self._token_store: return None data = await self._token_store.pop_token(RedisTokenStore.PREFIX_TOTP_RECOVERY, token) if not data: return None user_id = UUID(data["user_id"]) user = await self._user_repo.get_by_id(user_id) if not user: return None if self._totp_service: await self._totp_service.disable_by_user_id(user_id) refresh_token = generate_refresh_token() refresh_hash = hash_token(refresh_token) expires_at = datetime.now(timezone.utc) + timedelta(days=settings.jwt_refresh_ttl_days) session = await self._session_repo.create( user_id=user.id, refresh_token_hash=refresh_hash, expires_at=expires_at, ) access_token = create_access_token(user.id, session.id) refresh_jwt = create_refresh_jwt(user.id, session.id) now = datetime.now(timezone.utc) tokens = TokenResponse( access_token=access_token, refresh_token=refresh_jwt, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ) return (user, tokens) async def send_totp_recovery_code(self, email: str) -> dict: """Send 6-digit TOTP recovery code (embedded app style). Returns {sent, code?, reason?}.""" user = await self._user_repo.get_by_email(email.lower()) if not user: return {"sent": False, "reason": "account_not_found"} if not await self._totp_service.is_enabled(user.id): return {"sent": False, "reason": "totp_not_enabled"} if not self._token_store: return {"sent": False, "reason": "recovery_unavailable"} code = generate_6digit_code() await self._token_store.store_token( RedisTokenStore.PREFIX_TOTP_RECOVERY_CODE, code, {"user_id": str(user.id)}, 900, ) if self._email_provider: from src.domain.auth.email_templates import build_totp_recovery_code_email subject, html, text = build_totp_recovery_code_email(code) await self._email_provider.send( to=email.lower(), subject=subject, html=html, text=text, ) return {"sent": True} async def verify_totp_recovery_code(self, code: str) -> Optional[tuple[User, TokenResponse]]: """Consume 6-digit recovery code, disable TOTP, return (user, tokens) or None.""" if not self._token_store: return None normalized = "".join(c for c in str(code).strip() if c.isdigit()) if len(normalized) != 6: return None data = await self._token_store.pop_token(RedisTokenStore.PREFIX_TOTP_RECOVERY_CODE, normalized) if not data: return None user_id = UUID(data["user_id"]) user = await self._user_repo.get_by_id(user_id) if not user: return None if self._totp_service: await self._totp_service.disable_by_user_id(user_id) refresh_token = generate_refresh_token() refresh_hash = hash_token(refresh_token) expires_at = datetime.now(timezone.utc) + timedelta(days=settings.jwt_refresh_ttl_days) session = await self._session_repo.create(user_id=user.id, refresh_token_hash=refresh_hash, expires_at=expires_at) access_token = create_access_token(user.id, session.id) refresh_jwt = create_refresh_jwt(user.id, session.id) return ( user, TokenResponse( access_token=access_token, refresh_token=refresh_jwt, expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=datetime.now(timezone.utc), ), )