from datetime import timedelta from urllib.parse import urlencode import secrets import httpx from fastapi.responses import RedirectResponse from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from app.config import settings from app.database import get_db from app.dependencies import ( generate_reset_token, get_current_user, reset_token_expiry, ) from app.models.user import User from app.schemas.user import ( ForgotPasswordRequest, LoginRequest, MessageResponse, ResetPasswordRequest, TokenResponse, UserCreate, UserResponse, UserUpdate, ChangePasswordRequest, ) from app.services.user_service import ( authenticate_user, create_password_reset_token, create_user, get_user_by_email, get_user_by_username, get_valid_reset_token, update_user, ) from app.utils.security import ( create_access_token, create_oauth_state, create_refresh_token, hash_password, verify_oauth_state, verify_password, ) router = APIRouter(prefix="/auth", tags=["auth"]) @router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED) def register(data: UserCreate, db: Session = Depends(get_db)) -> User: if get_user_by_email(db, data.email.lower()): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Email already registered") if get_user_by_username(db, data.username.lower()): raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Username already taken") return create_user(db, data) @router.post("/login", response_model=TokenResponse) def login(data: LoginRequest, db: Session = Depends(get_db)) -> TokenResponse: user = authenticate_user(db, data.username_or_email, data.password) if user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username/email or password", ) if not user.is_active: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Account is inactive") access_token = create_access_token( user.email, timedelta(days=7) if data.remember_me else None, ) return TokenResponse( access_token=access_token, refresh_token=create_refresh_token(user.email), user=UserResponse.model_validate(user), ) @router.get("/me", response_model=UserResponse) def get_me(current_user: User = Depends(get_current_user)) -> User: return current_user @router.patch("/me", response_model=UserResponse) def update_me( data: UserUpdate, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> User: return update_user(db, current_user, data) @router.post("/forgot-password", response_model=MessageResponse) def forgot_password(data: ForgotPasswordRequest, db: Session = Depends(get_db)) -> MessageResponse: user = get_user_by_email(db, data.email.lower()) if user is None: return MessageResponse(message="If that email exists, a reset link has been sent.") token = generate_reset_token() create_password_reset_token(db, user, token, reset_token_expiry()) response = MessageResponse(message="If that email exists, a reset link has been sent.") if settings.debug: response.reset_token = token return response @router.post("/reset-password", response_model=MessageResponse) def reset_password(data: ResetPasswordRequest, db: Session = Depends(get_db)) -> MessageResponse: reset = get_valid_reset_token(db, data.token) if reset is None: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid or expired reset token") user = db.query(User).filter(User.id == reset.user_id).first() if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") user.hashed_password = hash_password(data.new_password) reset.used = True db.commit() return MessageResponse(message="Password reset successfully. You can now log in.") @router.post("/change-password", response_model=MessageResponse) def change_password( data: ChangePasswordRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> MessageResponse: if not verify_password(data.current_password, current_user.hashed_password): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Current password is incorrect") current_user.hashed_password = hash_password(data.new_password) db.commit() return MessageResponse(message="Password updated successfully.") @router.get("/google/login") def google_login() -> RedirectResponse: if not settings.google_client_id or not settings.google_client_secret: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Google sign-in is not configured. Set GOOGLE_CLIENT_ID and GOOGLE_CLIENT_SECRET.", ) state = create_oauth_state() query = urlencode({ "client_id": settings.google_client_id, "redirect_uri": settings.google_redirect_uri, "response_type": "code", "scope": "openid email profile", "state": state, "access_type": "offline", "prompt": "select_account", }) return RedirectResponse(f"https://accounts.google.com/o/oauth2/v2/auth?{query}") @router.get("/google/callback") async def google_callback(code: str, state: str | None = None) -> RedirectResponse: if not settings.google_client_id or not settings.google_client_secret: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Google sign-in is not configured") if not code or not state or not verify_oauth_state(state): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Google authorization was incomplete") async with httpx.AsyncClient(timeout=10) as client: token_response = await client.post( "https://oauth2.googleapis.com/token", data={ "code": code, "client_id": settings.google_client_id, "client_secret": settings.google_client_secret, "redirect_uri": settings.google_redirect_uri, "grant_type": "authorization_code", }, ) if token_response.status_code != 200: raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Google authorization failed") token_data = token_response.json() profile_response = await client.get( "https://openidconnect.googleapis.com/v1/userinfo", headers={"Authorization": f"Bearer {token_data['access_token']}"}, ) profile = profile_response.json() email = str(profile.get("email", "")).lower() if not email or not profile.get("email_verified"): raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Google account email is not verified") from app.services.user_service import get_user_by_email db = next(get_db()) try: user = get_user_by_email(db, email) if user is None: base_username = email.split("@")[0][:90] username = base_username suffix = 1 while get_user_by_username(db, username): username = f"{base_username}{suffix}" suffix += 1 user = create_user(db, UserCreate( email=email, username=username, password=secrets.token_urlsafe(32), full_name=str(profile.get("name") or username), )) access_token = create_access_token(user.email) refresh_token = create_refresh_token(user.email) finally: db.close() query = urlencode({"access_token": access_token, "refresh_token": refresh_token}) return RedirectResponse(f"{settings.frontend_url}/auth/google/callback?{query}")