"""Desktop-auth application service: login start/complete/exchange and browser handoff.""" from __future__ import annotations import secrets from datetime import datetime, timedelta, timezone from typing import Any, Optional from uuid import UUID from src.config import get_settings from src.core.exceptions import ForbiddenError, NotFoundError, UnauthorizedError, ValidationError from src.core.security.hashing import digest_token_for_lookup, generate_refresh_token, hash_token from src.core.security.jwt import ( DESKTOP_SESSION_TTL_MINUTES, create_access_token, create_desktop_session_token, create_refresh_token as create_refresh_jwt, ) from src.domain.auth.models import User from src.domain.auth.repository import SessionRepository, UserRepository from src.domain.auth.schemas import TokenResponse from src.domain.desktop_auth.handoff_allowlist import normalize_handoff_next from src.domain.desktop_auth.models import DesktopAuthStatus from src.domain.desktop_auth.repository import DesktopAuthRequestRepository, DesktopHandoffTokenRepository AUTH_TTL_MINUTES = 15 HANDOFF_TTL_SECONDS = 90 _USER_CODE_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" def _generate_login_code() -> str: return "dl_" + secrets.token_urlsafe(32) def _generate_user_code() -> str: return "".join(secrets.choice(_USER_CODE_ALPHABET) for _ in range(8)) def _generate_handoff_token() -> str: return "dho_" + secrets.token_urlsafe(32) class DesktopAuthService: def __init__( self, *, requests: DesktopAuthRequestRepository, handoffs: DesktopHandoffTokenRepository, users: UserRepository, sessions: SessionRepository, app_url: str, ) -> None: self._requests = requests self._handoffs = handoffs self._users = users self._sessions = sessions self._app_url = app_url.rstrip("/") async def start(self, *, app_name: Optional[str] = None, os: Optional[str] = None) -> dict[str, Any]: login_code = _generate_login_code() user_code = _generate_user_code() expires_at = datetime.now(timezone.utc) + timedelta(minutes=AUTH_TTL_MINUTES) row = await self._requests.create( login_code=login_code, user_code=user_code, expires_at=expires_at, app_name=app_name, os=os, ) return { "auth_id": row.id, "login_code": login_code, "user_code": user_code, "browser_url": f"{self._app_url}/desktop/login?code={user_code}", "expires_at": expires_at, } async def complete(self, *, user_code: str, user_id: UUID) -> dict[str, Any]: row = await self._requests.get_by_user_code(user_code.strip().upper()) if not row: # user codes are stored uppercase; also try the raw value row = await self._requests.get_by_user_code(user_code.strip()) if not row: raise NotFoundError("Desktop login code not found") if row.status == DesktopAuthStatus.exchanged.value: raise ValidationError("This desktop login has already been used") if row.expires_at < datetime.now(timezone.utc): raise ValidationError("Desktop login code has expired; start again from the app") if row.status == DesktopAuthStatus.completed.value: if row.user_id != user_id: raise ForbiddenError("This desktop login was completed by another account") return {"status": "completed", "auth_id": row.id} if row.status != DesktopAuthStatus.pending.value: raise ValidationError(f"Desktop login is not awaiting completion (status: {row.status})") claimed = await self._requests.mark_completed_if_pending(row.id, user_id) if not claimed: row = await self._requests.get_by_login_code(row.login_code) if not row: raise NotFoundError("Desktop login request not found") if row.status == DesktopAuthStatus.completed.value and row.user_id == user_id: return {"status": "completed", "auth_id": row.id} if row.status == DesktopAuthStatus.completed.value and row.user_id != user_id: raise ForbiddenError("This desktop login was completed by another account") if row.status == DesktopAuthStatus.exchanged.value: raise ValidationError("This desktop login has already been used") raise ValidationError(f"Desktop login is not awaiting completion (status: {row.status})") return {"status": "completed", "auth_id": row.id} async def exchange(self, *, login_code: str) -> dict[str, Any]: row = await self._requests.get_by_login_code(login_code) if not row: raise NotFoundError("Desktop login request not found") now = datetime.now(timezone.utc) if row.expires_at < now and row.status == DesktopAuthStatus.pending.value: return {"status": "expired"} if row.status == DesktopAuthStatus.pending.value: return {"status": "pending"} if row.status == DesktopAuthStatus.exchanged.value: return {"status": "exchanged"} if row.status != DesktopAuthStatus.completed.value or not row.user_id: return {"status": "pending"} if not await self._requests.mark_exchanged_if_completed(row.id): latest = await self._requests.get_by_login_code(login_code) if latest and latest.status == DesktopAuthStatus.exchanged.value: return {"status": "exchanged"} return {"status": "pending"} session = create_desktop_session_token(row.user_id) return { "status": "completed", "desktop_session": session, "expires_in": DESKTOP_SESSION_TTL_MINUTES * 60, "user_id": row.user_id, } async def mint_handoff(self, *, user_id: UUID, next_path: str | None) -> dict[str, Any]: try: dest = normalize_handoff_next(next_path) except ValueError as exc: raise ValidationError(str(exc)) from exc raw = _generate_handoff_token() expires_at = datetime.now(timezone.utc) + timedelta(seconds=HANDOFF_TTL_SECONDS) await self._handoffs.create( token_hash=digest_token_for_lookup(raw), user_id=user_id, next_path=dest, expires_at=expires_at, ) url = f"{self._app_url}/desktop/handoff?token={raw}&next={dest}" return {"url": url, "expires_in": HANDOFF_TTL_SECONDS} async def consume_handoff(self, *, token: str) -> tuple[TokenResponse, User, str]: """Redeem a one-time handoff token into a normal web session (access + refresh).""" row = await self._handoffs.get_by_token_hash(digest_token_for_lookup(token)) if not row: raise UnauthorizedError("Invalid or expired handoff token") now = datetime.now(timezone.utc) if row.used_at is not None: raise UnauthorizedError("Handoff token has already been used") if row.expires_at < now: raise UnauthorizedError("Handoff token has expired") user = await self._users.get_by_id(row.user_id) if not user: raise NotFoundError("User not found") if not await self._handoffs.mark_used_if_unused(row.id, now=now): raise UnauthorizedError("Handoff token has already been used") settings = get_settings() refresh_token = generate_refresh_token() expires_at = now + timedelta(days=settings.jwt_refresh_ttl_days) session = await self._sessions.create( user_id=user.id, refresh_token_hash=hash_token(refresh_token), expires_at=expires_at, device_info={"source": "desktop_handoff"}, ) tokens = TokenResponse( access_token=create_access_token(user.id, session.id), refresh_token=create_refresh_jwt(user.id, session.id), expires_in=settings.jwt_access_ttl_minutes * 60, issued_at=now, ) return tokens, user, row.next_path