File size: 6,476 Bytes
1941764 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 | """
Password Reset API endpoints for secure password recovery.
This module provides:
- POST /api/auth/forgot-password - Request password reset email
- GET /api/auth/reset-password/{token} - Verify reset token validity
- POST /api/auth/reset-password - Reset password with token
"""
from fastapi import APIRouter, HTTPException, Depends
from sqlmodel import Session
from pydantic import BaseModel, EmailStr, Field
from typing import Optional
from ..models.user import User
from ..services.auth import hash_password
from ..services.password_reset import (
create_reset_token,
validate_reset_token,
invalidate_token,
check_rate_limit,
validate_password_strength,
get_user_by_email
)
from ..services.email import send_password_reset_email
from ..database import get_session
router = APIRouter()
# Request/Response Models
class ForgotPasswordRequest(BaseModel):
"""Request model for forgot password."""
email: EmailStr = Field(..., description="User email address")
class ForgotPasswordResponse(BaseModel):
"""Response model for forgot password request."""
message: str
class TokenValidationResponse(BaseModel):
"""Response model for token validation."""
valid: bool
email: Optional[str] = None
error: Optional[str] = None
class ResetPasswordRequest(BaseModel):
"""Request model for password reset."""
token: str = Field(..., description="Password reset token")
new_password: str = Field(..., min_length=8, description="New password (minimum 8 characters)")
class ResetPasswordResponse(BaseModel):
"""Response model for password reset."""
message: str
@router.post("/forgot-password", response_model=ForgotPasswordResponse)
async def forgot_password(
request: ForgotPasswordRequest,
session: Session = Depends(get_session)
) -> ForgotPasswordResponse:
"""
Request a password reset email.
Security features:
- No user enumeration (same response for existing/non-existing emails)
- Rate limiting (3 requests per hour per user)
- Cryptographically secure tokens
- 15-minute token expiry
Args:
request: Forgot password request with email
session: Database session
Returns:
Generic success message (no user enumeration)
Raises:
HTTPException 400: If email format is invalid
HTTPException 429: If rate limit exceeded
"""
# Find user by email
user = get_user_by_email(session, request.email)
# Always return same message to prevent user enumeration
generic_message = "If an account exists with this email, you will receive a password reset link shortly."
# If user doesn't exist, return generic message (no enumeration)
if not user:
return ForgotPasswordResponse(message=generic_message)
# Check rate limit
if not check_rate_limit(session, user.id):
raise HTTPException(
status_code=429,
detail="Too many password reset requests. Please try again later."
)
# Create reset token
token = create_reset_token(session, user.id)
# Send reset email
email_sent = send_password_reset_email(user.email, token)
if not email_sent:
# Log error but don't expose to user
print(f"Failed to send password reset email to {user.email}")
# Always return generic message
return ForgotPasswordResponse(message=generic_message)
@router.get("/reset-password/{token}", response_model=TokenValidationResponse)
async def verify_reset_token(
token: str,
session: Session = Depends(get_session)
) -> TokenValidationResponse:
"""
Verify if a password reset token is valid.
Checks:
- Token exists
- Token has not expired (15 minutes)
- Token has not been used
Args:
token: Password reset token to verify
session: Database session
Returns:
TokenValidationResponse with validity status and user email
Example:
GET /api/auth/reset-password/abc123def456
"""
# Validate token
token_record = validate_reset_token(session, token)
if not token_record:
return TokenValidationResponse(
valid=False,
error="Invalid or expired reset token"
)
# Get user email
user = session.get(User, token_record.user_id)
if not user:
return TokenValidationResponse(
valid=False,
error="User not found"
)
return TokenValidationResponse(
valid=True,
email=user.email
)
@router.post("/reset-password", response_model=ResetPasswordResponse)
async def reset_password(
request: ResetPasswordRequest,
session: Session = Depends(get_session)
) -> ResetPasswordResponse:
"""
Reset user password with a valid token.
Security features:
- Token validation (expiry, usage)
- Password strength validation
- One-time use tokens
- Automatic token invalidation
Args:
request: Reset password request with token and new password
session: Database session
Returns:
Success message
Raises:
HTTPException 400: If token is invalid or password is weak
HTTPException 422: If validation fails
"""
# Validate token
token_record = validate_reset_token(session, request.token)
if not token_record:
raise HTTPException(
status_code=400,
detail="Invalid or expired reset token"
)
# Validate password strength
password_validation = validate_password_strength(request.new_password)
if not password_validation["valid"]:
raise HTTPException(
status_code=400,
detail={
"message": "Password does not meet strength requirements",
"errors": password_validation["errors"]
}
)
# Get user
user = session.get(User, token_record.user_id)
if not user:
raise HTTPException(
status_code=400,
detail="User not found"
)
# Hash new password
hashed_password = hash_password(request.new_password)
# Update user password
user.hashed_password = hashed_password
session.add(user)
# Invalidate token (mark as used)
invalidate_token(session, request.token)
# Commit changes
session.commit()
return ResetPasswordResponse(
message="Password successfully reset. You can now sign in with your new password."
)
|