Spaces:
Running
Running
| """ | |
| Mobile Authentication Routes | |
| Provides mobile-specific authentication endpoints: | |
| - Mobile login with device registration | |
| - Biometric authentication registration | |
| - Mobile token refresh | |
| - Device management | |
| """ | |
| import logging | |
| from datetime import datetime | |
| from typing import Any, Dict, Optional | |
| from fastapi import Depends, HTTPException, Request, status | |
| from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm | |
| from pydantic import BaseModel | |
| from sqlalchemy.orm import Session | |
| from core.auth import ( | |
| authenticate_mobile_user, | |
| create_access_token, | |
| create_mobile_token, | |
| get_current_user, | |
| get_mobile_device, | |
| verify_biometric_signature, | |
| verify_password, | |
| ) | |
| from core.base_routes import BaseAPIRouter | |
| from core.database import get_db | |
| from core.models import MobileDevice, User | |
| logger = logging.getLogger(__name__) | |
| router = BaseAPIRouter(prefix="/api/auth", tags=["Authentication"]) | |
| # ============================================================================ | |
| # Request/Response Models | |
| # ============================================================================ | |
| class MobileLoginRequest(BaseModel): | |
| email: str | |
| password: str | |
| device_token: str | |
| platform: str # ios, android | |
| device_info: Optional[Dict[str, Any]] = None | |
| class MobileLoginResponse(BaseModel): | |
| access_token: str | |
| refresh_token: str | |
| expires_at: str | |
| token_type: str | |
| user: Dict[str, Any] | |
| class BiometricRegisterRequest(BaseModel): | |
| public_key: str | |
| device_token: str | |
| platform: str | |
| class BiometricRegisterResponse(BaseModel): | |
| success: bool | |
| challenge: str | |
| message: str | |
| class BiometricAuthRequest(BaseModel): | |
| device_id: str | |
| signature: str | |
| challenge: str | |
| class BiometricAuthResponse(BaseModel): | |
| success: bool | |
| access_token: Optional[str] = None | |
| refresh_token: Optional[str] = None | |
| message: str | |
| class RefreshTokenRequest(BaseModel): | |
| refresh_token: str | |
| class DeviceInfoResponse(BaseModel): | |
| device_id: str | |
| platform: str | |
| status: str | |
| notification_enabled: bool | |
| last_active: str | |
| created_at: str | |
| # ============================================================================ | |
| # Mobile Authentication Routes | |
| # ============================================================================ | |
| async def mobile_login( | |
| request: MobileLoginRequest, | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Mobile login with automatic device registration. | |
| Args: | |
| request: Login credentials and device information | |
| db: Database session | |
| Returns: | |
| Access token, refresh token, and user information | |
| Raises: | |
| 401: Invalid credentials | |
| 400: Invalid request data | |
| """ | |
| try: | |
| # Authenticate user | |
| result = await authenticate_mobile_user( | |
| email=request.email, | |
| password=request.password, | |
| device_token=request.device_token, | |
| platform=request.platform, | |
| db=db | |
| ) | |
| if not result: | |
| raise router.validation_error( | |
| "credentials", | |
| "Invalid email or password" | |
| ) | |
| # Update device info if provided | |
| if request.device_info and result.get("user"): | |
| user_id = result["user"]["id"] | |
| device = db.query(MobileDevice).filter( | |
| MobileDevice.device_token == request.device_token | |
| ).first() | |
| if device: | |
| device.device_info = request.device_info | |
| device.last_active = datetime.utcnow() | |
| db.commit() | |
| logger.info(f"Mobile login successful for {request.email}") | |
| return MobileLoginResponse(**result) | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Mobile login error: {e}") | |
| raise router.internal_error(f"Login failed: {str(e)}") | |
| async def register_biometric( | |
| request: BiometricRegisterRequest, | |
| current_user: User = Depends(get_current_user), | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Register device for biometric authentication (Face ID, Touch ID). | |
| Args: | |
| request: Biometric registration data | |
| current_user: Authenticated user | |
| db: Database session | |
| Returns: | |
| Challenge string for device to sign | |
| Raises: | |
| 400: Invalid request | |
| 404: Device not found | |
| """ | |
| try: | |
| # Find device by token | |
| device = db.query(MobileDevice).filter( | |
| MobileDevice.device_token == request.device_token, | |
| MobileDevice.user_id == str(current_user.id) | |
| ).first() | |
| if not device: | |
| raise router.not_found_error("Device", request.device_token) | |
| # Generate challenge | |
| import secrets | |
| challenge = secrets.token_urlsafe(32) | |
| # Store public key (in production, this should be encrypted) | |
| # For now, we'll store it in device_info | |
| device_info = device.device_info or {} | |
| device_info["biometric_public_key"] = request.public_key | |
| device_info["biometric_challenge"] = challenge | |
| device_info["biometric_enabled"] = False # Will be enabled after first successful auth | |
| device.device_info = device_info | |
| device.last_active = datetime.utcnow() | |
| db.commit() | |
| logger.info(f"Biometric registration initiated for device {device.id}") | |
| return BiometricRegisterResponse( | |
| success=True, | |
| challenge=challenge, | |
| message="Biometric registration initiated. Please sign the challenge." | |
| ) | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Biometric registration error: {e}") | |
| raise router.internal_error(f"Registration failed: {str(e)}") | |
| async def authenticate_with_biometric( | |
| request: BiometricAuthRequest, | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Authenticate using biometric signature. | |
| Args: | |
| request: Biometric authentication data | |
| db: Database session | |
| Returns: | |
| Access tokens if authentication successful | |
| Raises: | |
| 401: Invalid signature | |
| 404: Device not found | |
| """ | |
| try: | |
| # Get device | |
| device = get_mobile_device(request.device_id, request.signature, db) | |
| if not device: | |
| # Try to get device by ID | |
| device = db.query(MobileDevice).filter( | |
| MobileDevice.id == request.device_id | |
| ).first() | |
| if not device: | |
| raise router.not_found_error("Device", request.device_id) | |
| # Get stored public key and challenge | |
| device_info = device.device_info or {} | |
| public_key = device_info.get("biometric_public_key") | |
| stored_challenge = device_info.get("biometric_challenge") | |
| if not public_key: | |
| raise router.validation_error( | |
| "biometric", | |
| "Biometric not registered for this device" | |
| ) | |
| # Verify signature | |
| if not verify_biometric_signature(request.signature, public_key, request.challenge): | |
| logger.warning(f"Biometric authentication failed for device {device.id}") | |
| return BiometricAuthResponse( | |
| success=False, | |
| message="Invalid signature" | |
| ) | |
| # Signature is valid, get user | |
| user = db.query(User).filter(User.id == device.user_id).first() | |
| if not user: | |
| raise router.not_found_error("User", device.user_id) | |
| # Generate tokens | |
| tokens = create_mobile_token(user, device.id) | |
| # Mark biometric as enabled | |
| device_info["biometric_enabled"] = True | |
| device_info["last_biometric_auth"] = datetime.utcnow().isoformat() | |
| device.device_info = device_info | |
| device.last_active = datetime.utcnow() | |
| db.commit() | |
| logger.info(f"Biometric authentication successful for user {user.email}") | |
| return BiometricAuthResponse( | |
| success=True, | |
| access_token=tokens["access_token"], | |
| refresh_token=tokens["refresh_token"], | |
| message="Authentication successful" | |
| ) | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Biometric authentication error: {e}") | |
| raise router.internal_error(f"Authentication failed: {str(e)}") | |
| async def refresh_mobile_token( | |
| request: RefreshTokenRequest, | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Refresh mobile access token using refresh token. | |
| Args: | |
| request: Refresh token | |
| db: Database session | |
| Returns: | |
| New access and refresh tokens | |
| Raises: | |
| 401: Invalid refresh token | |
| """ | |
| try: | |
| from jose import JWTError, jwt | |
| # Decode refresh token | |
| try: | |
| payload = jwt.decode( | |
| request.refresh_token, | |
| router.auth_module.SECRET_KEY if hasattr(router, 'auth_module') else os.getenv("SECRET_KEY"), | |
| algorithms=["HS256"] | |
| ) | |
| except JWTError: | |
| raise router.validation_error("token", "Invalid refresh token") | |
| user_id = payload.get("sub") | |
| token_type = payload.get("type") | |
| device_id = payload.get("device_id") | |
| if not user_id or token_type != "refresh": | |
| raise router.validation_error("token", "Invalid refresh token") | |
| # Get user | |
| user = db.query(User).filter(User.id == user_id).first() | |
| if not user: | |
| raise router.not_found_error("User", user_id) | |
| # Verify device exists and is active | |
| device = get_mobile_device(device_id, user_id, db) | |
| if not device: | |
| raise router.validation_error("device", "Device not found or inactive") | |
| # Generate new tokens | |
| tokens = create_mobile_token(user, device_id) | |
| logger.info(f"Token refresh successful for user {user.email}") | |
| return tokens | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Token refresh error: {e}") | |
| raise router.internal_error(f"Refresh failed: {str(e)}") | |
| async def get_mobile_device_info( | |
| device_id: str, | |
| current_user: User = Depends(get_current_user), | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Get mobile device information. | |
| Args: | |
| device_id: Device ID | |
| current_user: Authenticated user | |
| db: Database session | |
| Returns: | |
| Device information | |
| Raises: | |
| 404: Device not found | |
| """ | |
| try: | |
| device = get_mobile_device(device_id, str(current_user.id), db) | |
| if not device: | |
| raise router.not_found_error("Device", device_id) | |
| return DeviceInfoResponse( | |
| device_id=device.id, | |
| platform=device.platform, | |
| status=device.status, | |
| notification_enabled=device.notification_enabled, | |
| last_active=device.last_active.isoformat(), | |
| created_at=device.created_at.isoformat() | |
| ) | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Get device info error: {e}") | |
| raise router.internal_error(f"Failed to get device info: {str(e)}") | |
| async def delete_mobile_device( | |
| device_id: str, | |
| current_user: User = Depends(get_current_user), | |
| db: Session = Depends(get_db) | |
| ): | |
| """ | |
| Unregister mobile device. | |
| Args: | |
| device_id: Device ID | |
| current_user: Authenticated user | |
| db: Database session | |
| Returns: | |
| Success message | |
| """ | |
| try: | |
| device = get_mobile_device(device_id, str(current_user.id), db) | |
| if not device: | |
| raise router.not_found_error("Device", device_id) | |
| # Mark as inactive instead of deleting | |
| device.status = "inactive" | |
| device.notification_enabled = False | |
| device.last_active = datetime.utcnow() | |
| db.commit() | |
| logger.info(f"Device {device_id} unregistered by user {current_user.email}") | |
| return router.success_response(message="Device unregistered successfully") | |
| except HTTPException: | |
| raise | |
| except Exception as e: | |
| logger.error(f"Delete device error: {e}") | |
| raise router.internal_error(f"Failed to unregister device: {str(e)}") | |