272 lines
8.8 KiB
Python
272 lines
8.8 KiB
Python
"""Auth API routes for VoIdea."""
|
|
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.dependencies import get_current_user, get_db
|
|
from app.core.limiter import limiter
|
|
from app.core.security import (
|
|
create_access_token,
|
|
create_refresh_token,
|
|
verify_password,
|
|
get_password_hash,
|
|
)
|
|
from app.integrations.oauth.yandex import exchange_code, get_authorize_url, get_user_info
|
|
from app.models.user import User
|
|
from app.schemas.auth import (
|
|
ForgotPasswordRequest,
|
|
LoginRequest,
|
|
OAuthCallbackRequest,
|
|
OAuthUrlResponse,
|
|
RefreshRequest,
|
|
RegisterRequest,
|
|
ResetPasswordRequest,
|
|
TokenResponse,
|
|
TwoFactorLoginRequest,
|
|
TwoFactorLoginResponse,
|
|
TwoFactorSetupResponse,
|
|
TwoFactorVerifyRequest,
|
|
)
|
|
from app.schemas.user import ChangePasswordRequest
|
|
from app.services.auth_service import AuthService
|
|
from app.services.password_reset_service import reset_password, send_reset_email
|
|
from app.services.two_factor_service import (
|
|
generate_qr_base64,
|
|
generate_totp_secret,
|
|
get_totp_uri,
|
|
get_user_secret,
|
|
is_2fa_enabled,
|
|
is_2fa_globally_enabled,
|
|
set_2fa_enabled,
|
|
set_user_secret,
|
|
verify_totp,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/register", response_model=TokenResponse, status_code=status.HTTP_201_CREATED)
|
|
@limiter.limit("5/minute")
|
|
async def register(request: Request, body: RegisterRequest, db: AsyncSession = Depends(get_db)):
|
|
service = AuthService(db)
|
|
try:
|
|
return await service.register(
|
|
body.email, body.password, body.display_name,
|
|
accepted_terms=body.accepted_terms,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
|
|
|
|
|
|
@router.post("/login")
|
|
@limiter.limit("10/minute")
|
|
async def login(request: Request, body: LoginRequest, db: AsyncSession = Depends(get_db)):
|
|
from sqlalchemy import select
|
|
result = await db.execute(select(User).where(User.email == body.email))
|
|
user = result.scalar_one_or_none()
|
|
if not user or not verify_password(body.password, user.password_hash):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверный email или пароль")
|
|
|
|
globally_enabled = await is_2fa_globally_enabled(db)
|
|
if globally_enabled and is_2fa_enabled(user):
|
|
from datetime import timedelta
|
|
temp_token = create_access_token(
|
|
data={"sub": str(user.id), "purpose": "2fa"},
|
|
expires_delta=timedelta(minutes=5),
|
|
)
|
|
return {"temp_token": temp_token, "message": "Требуется 2FA код"}
|
|
|
|
service = AuthService(db)
|
|
return await service.create_token_response(user)
|
|
|
|
|
|
@router.post("/2fa/verify-login", response_model=TwoFactorLoginResponse)
|
|
@limiter.limit("10/minute")
|
|
async def verify_2fa_login(
|
|
request: Request,
|
|
body: TwoFactorLoginRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
from jose import jwt, JWTError
|
|
from app.core.config import get_settings
|
|
settings = get_settings()
|
|
|
|
try:
|
|
payload = jwt.decode(
|
|
body.temp_token, settings.jwt_secret_key,
|
|
algorithms=[settings.jwt_algorithm],
|
|
)
|
|
except JWTError:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid temp token")
|
|
|
|
if payload.get("purpose") != "2fa":
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token purpose")
|
|
|
|
from sqlalchemy import select
|
|
result = await db.execute(select(User).where(User.id == payload.get("sub")))
|
|
user = result.scalar_one_or_none()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
|
|
|
|
secret = get_user_secret(user)
|
|
if not secret or not verify_totp(secret, body.totp_code):
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Неверный 2FA код")
|
|
|
|
service = AuthService(db)
|
|
return await service.create_token_response(user)
|
|
|
|
|
|
@router.post("/2fa/setup", response_model=TwoFactorSetupResponse)
|
|
async def setup_2fa(
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if is_2fa_enabled(user):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA уже включена")
|
|
|
|
secret = generate_totp_secret()
|
|
set_user_secret(user, secret)
|
|
await db.commit()
|
|
|
|
uri = get_totp_uri(secret, user.email)
|
|
qr = generate_qr_base64(uri)
|
|
|
|
return TwoFactorSetupResponse(secret=secret, uri=uri, qr_base64=qr)
|
|
|
|
|
|
@router.post("/2fa/verify")
|
|
async def verify_2fa(
|
|
body: TwoFactorVerifyRequest,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
secret = get_user_secret(user)
|
|
if not secret:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="2FA не настроена")
|
|
|
|
if not verify_totp(secret, body.token):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Неверный код")
|
|
|
|
set_2fa_enabled(user, True)
|
|
await db.commit()
|
|
return {"message": "2FA успешно включена"}
|
|
|
|
|
|
@router.post("/2fa/disable")
|
|
async def disable_2fa(
|
|
body: TwoFactorVerifyRequest,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
secret = get_user_secret(user)
|
|
if not secret or not verify_totp(secret, body.token):
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Неверный код")
|
|
|
|
set_2fa_enabled(user, False)
|
|
await db.commit()
|
|
return {"message": "2FA отключена"}
|
|
|
|
|
|
@router.post("/refresh", response_model=TokenResponse)
|
|
@limiter.limit("10/minute")
|
|
async def refresh(request: Request, body: RefreshRequest, db: AsyncSession = Depends(get_db)):
|
|
service = AuthService(db)
|
|
try:
|
|
return await service.refresh(body.refresh_token)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(e))
|
|
|
|
|
|
@router.get("/oauth/yandex", response_model=OAuthUrlResponse)
|
|
@limiter.limit("10/minute")
|
|
async def oauth_yandex_url(request: Request):
|
|
url = await get_authorize_url()
|
|
return OAuthUrlResponse(url=url, provider="yandex")
|
|
|
|
|
|
@router.post("/oauth/yandex/callback", response_model=TokenResponse)
|
|
@limiter.limit("10/minute")
|
|
async def oauth_yandex_callback(
|
|
request: Request,
|
|
body: OAuthCallbackRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
token_result = await exchange_code(body.code)
|
|
if not token_result:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Failed to exchange authorization code",
|
|
)
|
|
|
|
user_info = await get_user_info(token_result.access_token)
|
|
if not user_info:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Failed to get user info from Yandex",
|
|
)
|
|
|
|
service = AuthService(db)
|
|
try:
|
|
return await service.oauth_or_register_login(
|
|
email=user_info.email,
|
|
oauth_provider="yandex",
|
|
oauth_id=user_info.id,
|
|
display_name=user_info.display_name,
|
|
avatar_url=user_info.avatar_url,
|
|
)
|
|
except ValueError as e:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))
|
|
|
|
|
|
@router.post("/forgot-password")
|
|
@limiter.limit("3/minute")
|
|
async def forgot_password(
|
|
request: Request,
|
|
body: ForgotPasswordRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
success, message = await send_reset_email(db, body.email)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
|
|
detail=message,
|
|
)
|
|
return {"message": message}
|
|
|
|
|
|
@router.post("/reset-password")
|
|
@limiter.limit("5/minute")
|
|
async def reset_password_endpoint(
|
|
request: Request,
|
|
body: ResetPasswordRequest,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
success, message = await reset_password(db, body.token, body.new_password)
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=message,
|
|
)
|
|
return {"message": message}
|
|
|
|
|
|
@router.post("/change-password")
|
|
@limiter.limit("5/minute")
|
|
async def change_password(
|
|
request: Request,
|
|
body: ChangePasswordRequest,
|
|
user: Annotated[User, Depends(get_current_user)],
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
if not verify_password(body.current_password, user.password_hash):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="Неверный текущий пароль",
|
|
)
|
|
|
|
user.password_hash = get_password_hash(body.new_password)
|
|
await db.commit()
|
|
return {"message": "Пароль успешно изменён"}
|