forked from cardosofelipe/fast-next-template
Add password reset functionality, email service, and related API endpoints
- Introduced endpoints for requesting and confirming password resets. - Implemented token-based password reset logic with validation checks. - Added `EmailService` with `ConsoleEmailBackend` and placeholder for SMTP backend. - Integrated password reset flow in `auth` API routes with rate limiting. - Updated schemas for password reset requests and token confirmation. - Refined validation for secure password updates and token verification. - Enhanced configuration with `FRONTEND_URL` for email links.
This commit is contained in:
Felipe Cardoso
@@ -17,9 +17,16 @@ from app.schemas.users import (
|
||||
UserResponse,
|
||||
Token,
|
||||
LoginRequest,
|
||||
RefreshTokenRequest
|
||||
RefreshTokenRequest,
|
||||
PasswordResetRequest,
|
||||
PasswordResetConfirm
|
||||
)
|
||||
from app.schemas.common import MessageResponse
|
||||
from app.services.auth_service import AuthService, AuthenticationError
|
||||
from app.services.email_service import email_service
|
||||
from app.utils.security import create_password_reset_token, verify_password_reset_token
|
||||
from app.crud.user import user as user_crud
|
||||
from app.core.auth import get_password_hash
|
||||
|
||||
router = APIRouter()
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -204,7 +211,139 @@ async def get_current_user_info(
|
||||
) -> Any:
|
||||
"""
|
||||
Get current user information.
|
||||
|
||||
|
||||
Requires authentication.
|
||||
"""
|
||||
return current_user
|
||||
|
||||
|
||||
@router.post(
|
||||
"/password-reset/request",
|
||||
response_model=MessageResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Request Password Reset",
|
||||
description="""
|
||||
Request a password reset link.
|
||||
|
||||
An email will be sent with a reset link if the email exists.
|
||||
Always returns success to prevent email enumeration.
|
||||
|
||||
**Rate Limit**: 3 requests/minute
|
||||
""",
|
||||
operation_id="request_password_reset"
|
||||
)
|
||||
@limiter.limit("3/minute")
|
||||
async def request_password_reset(
|
||||
request: Request,
|
||||
reset_request: PasswordResetRequest,
|
||||
db: Session = Depends(get_db)
|
||||
) -> Any:
|
||||
"""
|
||||
Request a password reset.
|
||||
|
||||
Sends an email with a password reset link.
|
||||
Always returns success to prevent email enumeration.
|
||||
"""
|
||||
try:
|
||||
# Look up user by email
|
||||
user = user_crud.get_by_email(db, email=reset_request.email)
|
||||
|
||||
# Only send email if user exists and is active
|
||||
if user and user.is_active:
|
||||
# Generate reset token
|
||||
reset_token = create_password_reset_token(user.email)
|
||||
|
||||
# Send password reset email
|
||||
await email_service.send_password_reset_email(
|
||||
to_email=user.email,
|
||||
reset_token=reset_token,
|
||||
user_name=user.first_name
|
||||
)
|
||||
logger.info(f"Password reset requested for {user.email}")
|
||||
else:
|
||||
# Log attempt but don't reveal if email exists
|
||||
logger.warning(f"Password reset requested for non-existent or inactive email: {reset_request.email}")
|
||||
|
||||
# Always return success to prevent email enumeration
|
||||
return MessageResponse(
|
||||
success=True,
|
||||
message="If your email is registered, you will receive a password reset link shortly"
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error processing password reset request: {str(e)}", exc_info=True)
|
||||
# Still return success to prevent information leakage
|
||||
return MessageResponse(
|
||||
success=True,
|
||||
message="If your email is registered, you will receive a password reset link shortly"
|
||||
)
|
||||
|
||||
|
||||
@router.post(
|
||||
"/password-reset/confirm",
|
||||
response_model=MessageResponse,
|
||||
status_code=status.HTTP_200_OK,
|
||||
summary="Confirm Password Reset",
|
||||
description="""
|
||||
Reset password using a token from email.
|
||||
|
||||
**Rate Limit**: 5 requests/minute
|
||||
""",
|
||||
operation_id="confirm_password_reset"
|
||||
)
|
||||
@limiter.limit("5/minute")
|
||||
def confirm_password_reset(
|
||||
request: Request,
|
||||
reset_confirm: PasswordResetConfirm,
|
||||
db: Session = Depends(get_db)
|
||||
) -> Any:
|
||||
"""
|
||||
Confirm password reset with token.
|
||||
|
||||
Verifies the token and updates the user's password.
|
||||
"""
|
||||
try:
|
||||
# Verify the reset token
|
||||
email = verify_password_reset_token(reset_confirm.token)
|
||||
|
||||
if not email:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Invalid or expired password reset token"
|
||||
)
|
||||
|
||||
# Look up user
|
||||
user = user_crud.get_by_email(db, email=email)
|
||||
|
||||
if not user:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail="User not found"
|
||||
)
|
||||
|
||||
if not user.is_active:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="User account is inactive"
|
||||
)
|
||||
|
||||
# Update password
|
||||
user.password_hash = get_password_hash(reset_confirm.new_password)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
|
||||
logger.info(f"Password reset successful for {user.email}")
|
||||
|
||||
return MessageResponse(
|
||||
success=True,
|
||||
message="Password has been reset successfully. You can now log in with your new password."
|
||||
)
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error confirming password reset: {str(e)}", exc_info=True)
|
||||
db.rollback()
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
||||
detail="An error occurred while resetting your password"
|
||||
)
|
||||
|
||||
@@ -58,6 +58,12 @@ class Settings(BaseSettings):
|
||||
# CORS configuration
|
||||
BACKEND_CORS_ORIGINS: List[str] = ["http://localhost:3000"]
|
||||
|
||||
# Frontend URL for email links
|
||||
FRONTEND_URL: str = Field(
|
||||
default="http://localhost:3000",
|
||||
description="Frontend application URL for email links"
|
||||
)
|
||||
|
||||
# Admin user
|
||||
FIRST_SUPERUSER_EMAIL: Optional[str] = Field(
|
||||
default=None,
|
||||
|
||||
@@ -4,7 +4,7 @@ from datetime import datetime
|
||||
from typing import Optional, Dict, Any
|
||||
from uuid import UUID
|
||||
|
||||
from pydantic import BaseModel, EmailStr, field_validator, ConfigDict
|
||||
from pydantic import BaseModel, EmailStr, field_validator, ConfigDict, Field
|
||||
|
||||
|
||||
class UserBase(BaseModel):
|
||||
@@ -166,3 +166,43 @@ class LoginRequest(BaseModel):
|
||||
|
||||
class RefreshTokenRequest(BaseModel):
|
||||
refresh_token: str
|
||||
|
||||
|
||||
class PasswordResetRequest(BaseModel):
|
||||
"""Schema for requesting a password reset."""
|
||||
email: EmailStr = Field(..., description="Email address of the account")
|
||||
|
||||
model_config = {
|
||||
"json_schema_extra": {
|
||||
"example": {
|
||||
"email": "user@example.com"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class PasswordResetConfirm(BaseModel):
|
||||
"""Schema for confirming a password reset with token."""
|
||||
token: str = Field(..., description="Password reset token from email")
|
||||
new_password: str = Field(..., min_length=8, description="New password")
|
||||
|
||||
@field_validator('new_password')
|
||||
@classmethod
|
||||
def password_strength(cls, v: str) -> str:
|
||||
"""Basic password strength validation"""
|
||||
if len(v) < 8:
|
||||
raise ValueError('Password must be at least 8 characters')
|
||||
if not any(char.isdigit() for char in v):
|
||||
raise ValueError('Password must contain at least one digit')
|
||||
if not any(char.isupper() for char in v):
|
||||
raise ValueError('Password must contain at least one uppercase letter')
|
||||
return v
|
||||
|
||||
model_config = {
|
||||
"json_schema_extra": {
|
||||
"example": {
|
||||
"token": "eyJwYXlsb2FkIjp7ImVtYWlsIjoidXNlckBleGFtcGxlLmNvbSIsImV4cCI6MTcxMjM0NTY3OH19",
|
||||
"new_password": "NewSecurePassword123"
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
300
backend/app/services/email_service.py
Normal file
300
backend/app/services/email_service.py
Normal file
@@ -0,0 +1,300 @@
|
||||
# app/services/email_service.py
|
||||
"""
|
||||
Email service with placeholder implementation.
|
||||
|
||||
This service provides email sending functionality with a simple console/log-based
|
||||
placeholder that can be easily replaced with a real email provider (SendGrid, SES, etc.)
|
||||
"""
|
||||
import logging
|
||||
from typing import List, Optional
|
||||
from abc import ABC, abstractmethod
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class EmailBackend(ABC):
|
||||
"""Abstract base class for email backends."""
|
||||
|
||||
@abstractmethod
|
||||
async def send_email(
|
||||
self,
|
||||
to: List[str],
|
||||
subject: str,
|
||||
html_content: str,
|
||||
text_content: Optional[str] = None
|
||||
) -> bool:
|
||||
"""Send an email."""
|
||||
pass
|
||||
|
||||
|
||||
class ConsoleEmailBackend(EmailBackend):
|
||||
"""
|
||||
Console/log-based email backend for development and testing.
|
||||
|
||||
This backend logs email content instead of actually sending emails.
|
||||
Replace this with a real backend (SMTP, SendGrid, SES) for production.
|
||||
"""
|
||||
|
||||
async def send_email(
|
||||
self,
|
||||
to: List[str],
|
||||
subject: str,
|
||||
html_content: str,
|
||||
text_content: Optional[str] = None
|
||||
) -> bool:
|
||||
"""
|
||||
Log email content to console/logs.
|
||||
|
||||
Args:
|
||||
to: List of recipient email addresses
|
||||
subject: Email subject
|
||||
html_content: HTML version of the email
|
||||
text_content: Plain text version of the email
|
||||
|
||||
Returns:
|
||||
True if "sent" successfully
|
||||
"""
|
||||
logger.info("=" * 80)
|
||||
logger.info("EMAIL SENT (Console Backend)")
|
||||
logger.info("=" * 80)
|
||||
logger.info(f"To: {', '.join(to)}")
|
||||
logger.info(f"Subject: {subject}")
|
||||
logger.info("-" * 80)
|
||||
if text_content:
|
||||
logger.info("Plain Text Content:")
|
||||
logger.info(text_content)
|
||||
logger.info("-" * 80)
|
||||
logger.info("HTML Content:")
|
||||
logger.info(html_content)
|
||||
logger.info("=" * 80)
|
||||
return True
|
||||
|
||||
|
||||
class SMTPEmailBackend(EmailBackend):
|
||||
"""
|
||||
SMTP email backend for production use.
|
||||
|
||||
TODO: Implement SMTP sending with proper error handling.
|
||||
This is a placeholder for future implementation.
|
||||
"""
|
||||
|
||||
def __init__(self, host: str, port: int, username: str, password: str):
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.username = username
|
||||
self.password = password
|
||||
|
||||
async def send_email(
|
||||
self,
|
||||
to: List[str],
|
||||
subject: str,
|
||||
html_content: str,
|
||||
text_content: Optional[str] = None
|
||||
) -> bool:
|
||||
"""Send email via SMTP."""
|
||||
# TODO: Implement SMTP sending
|
||||
logger.warning("SMTP backend not yet implemented, falling back to console")
|
||||
console_backend = ConsoleEmailBackend()
|
||||
return await console_backend.send_email(to, subject, html_content, text_content)
|
||||
|
||||
|
||||
class EmailService:
|
||||
"""
|
||||
High-level email service that uses different backends.
|
||||
|
||||
This service provides a clean interface for sending various types of emails
|
||||
and can be configured to use different backends (console, SMTP, SendGrid, etc.)
|
||||
"""
|
||||
|
||||
def __init__(self, backend: Optional[EmailBackend] = None):
|
||||
"""
|
||||
Initialize email service with a backend.
|
||||
|
||||
Args:
|
||||
backend: Email backend to use. Defaults to ConsoleEmailBackend.
|
||||
"""
|
||||
self.backend = backend or ConsoleEmailBackend()
|
||||
|
||||
async def send_password_reset_email(
|
||||
self,
|
||||
to_email: str,
|
||||
reset_token: str,
|
||||
user_name: Optional[str] = None
|
||||
) -> bool:
|
||||
"""
|
||||
Send password reset email.
|
||||
|
||||
Args:
|
||||
to_email: Recipient email address
|
||||
reset_token: Password reset token
|
||||
user_name: User's name for personalization
|
||||
|
||||
Returns:
|
||||
True if email sent successfully
|
||||
"""
|
||||
# Generate reset URL
|
||||
reset_url = f"{settings.FRONTEND_URL}/reset-password?token={reset_token}"
|
||||
|
||||
# Prepare email content
|
||||
subject = "Password Reset Request"
|
||||
|
||||
# Plain text version
|
||||
text_content = f"""
|
||||
Hello{' ' + user_name if user_name else ''},
|
||||
|
||||
You requested a password reset for your account. Click the link below to reset your password:
|
||||
|
||||
{reset_url}
|
||||
|
||||
This link will expire in 1 hour.
|
||||
|
||||
If you didn't request this, please ignore this email.
|
||||
|
||||
Best regards,
|
||||
The {settings.PROJECT_NAME} Team
|
||||
"""
|
||||
|
||||
# HTML version
|
||||
html_content = f"""
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<style>
|
||||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||||
.header {{ background-color: #4CAF50; color: white; padding: 20px; text-align: center; }}
|
||||
.content {{ padding: 20px; background-color: #f9f9f9; }}
|
||||
.button {{ display: inline-block; padding: 12px 24px; background-color: #4CAF50;
|
||||
color: white; text-decoration: none; border-radius: 4px; margin: 20px 0; }}
|
||||
.footer {{ padding: 20px; text-align: center; color: #777; font-size: 12px; }}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="container">
|
||||
<div class="header">
|
||||
<h1>Password Reset</h1>
|
||||
</div>
|
||||
<div class="content">
|
||||
<p>Hello{' ' + user_name if user_name else ''},</p>
|
||||
<p>You requested a password reset for your account. Click the button below to reset your password:</p>
|
||||
<p style="text-align: center;">
|
||||
<a href="{reset_url}" class="button">Reset Password</a>
|
||||
</p>
|
||||
<p>Or copy and paste this link into your browser:</p>
|
||||
<p style="word-break: break-all; color: #4CAF50;">{reset_url}</p>
|
||||
<p><strong>This link will expire in 1 hour.</strong></p>
|
||||
<p>If you didn't request this, please ignore this email.</p>
|
||||
</div>
|
||||
<div class="footer">
|
||||
<p>Best regards,<br>The {settings.PROJECT_NAME} Team</p>
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
try:
|
||||
return await self.backend.send_email(
|
||||
to=[to_email],
|
||||
subject=subject,
|
||||
html_content=html_content,
|
||||
text_content=text_content
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send password reset email to {to_email}: {str(e)}")
|
||||
return False
|
||||
|
||||
async def send_email_verification(
|
||||
self,
|
||||
to_email: str,
|
||||
verification_token: str,
|
||||
user_name: Optional[str] = None
|
||||
) -> bool:
|
||||
"""
|
||||
Send email verification email.
|
||||
|
||||
Args:
|
||||
to_email: Recipient email address
|
||||
verification_token: Email verification token
|
||||
user_name: User's name for personalization
|
||||
|
||||
Returns:
|
||||
True if email sent successfully
|
||||
"""
|
||||
# Generate verification URL
|
||||
verification_url = f"{settings.FRONTEND_URL}/verify-email?token={verification_token}"
|
||||
|
||||
# Prepare email content
|
||||
subject = "Verify Your Email Address"
|
||||
|
||||
# Plain text version
|
||||
text_content = f"""
|
||||
Hello{' ' + user_name if user_name else ''},
|
||||
|
||||
Thank you for signing up! Please verify your email address by clicking the link below:
|
||||
|
||||
{verification_url}
|
||||
|
||||
This link will expire in 24 hours.
|
||||
|
||||
If you didn't create an account, please ignore this email.
|
||||
|
||||
Best regards,
|
||||
The {settings.PROJECT_NAME} Team
|
||||
"""
|
||||
|
||||
# HTML version
|
||||
html_content = f"""
|
||||
<!DOCTYPE html>
|
||||
<html>
|
||||
<head>
|
||||
<style>
|
||||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||||
.header {{ background-color: #2196F3; color: white; padding: 20px; text-align: center; }}
|
||||
.content {{ padding: 20px; background-color: #f9f9f9; }}
|
||||
.button {{ display: inline-block; padding: 12px 24px; background-color: #2196F3;
|
||||
color: white; text-decoration: none; border-radius: 4px; margin: 20px 0; }}
|
||||
.footer {{ padding: 20px; text-align: center; color: #777; font-size: 12px; }}
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="container">
|
||||
<div class="header">
|
||||
<h1>Verify Your Email</h1>
|
||||
</div>
|
||||
<div class="content">
|
||||
<p>Hello{' ' + user_name if user_name else ''},</p>
|
||||
<p>Thank you for signing up! Please verify your email address by clicking the button below:</p>
|
||||
<p style="text-align: center;">
|
||||
<a href="{verification_url}" class="button">Verify Email</a>
|
||||
</p>
|
||||
<p>Or copy and paste this link into your browser:</p>
|
||||
<p style="word-break: break-all; color: #2196F3;">{verification_url}</p>
|
||||
<p><strong>This link will expire in 24 hours.</strong></p>
|
||||
<p>If you didn't create an account, please ignore this email.</p>
|
||||
</div>
|
||||
<div class="footer">
|
||||
<p>Best regards,<br>The {settings.PROJECT_NAME} Team</p>
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
"""
|
||||
|
||||
try:
|
||||
return await self.backend.send_email(
|
||||
to=[to_email],
|
||||
subject=subject,
|
||||
html_content=html_content,
|
||||
text_content=text_content
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to send verification email to {to_email}: {str(e)}")
|
||||
return False
|
||||
|
||||
|
||||
# Global email service instance
|
||||
email_service = EmailService()
|
||||
@@ -11,6 +11,7 @@ import json
|
||||
import secrets
|
||||
import time
|
||||
from typing import Dict, Any, Optional
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
@@ -108,3 +109,189 @@ def verify_upload_token(token: str) -> Optional[Dict[str, Any]]:
|
||||
|
||||
except (ValueError, KeyError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def create_password_reset_token(email: str, expires_in: int = 3600) -> str:
|
||||
"""
|
||||
Create a signed token for password reset.
|
||||
|
||||
Args:
|
||||
email: User's email address
|
||||
expires_in: Expiration time in seconds (default: 3600 = 1 hour)
|
||||
|
||||
Returns:
|
||||
A base64 encoded token string
|
||||
|
||||
Example:
|
||||
>>> token = create_password_reset_token("user@example.com")
|
||||
>>> # Send token to user via email
|
||||
"""
|
||||
# Create the payload
|
||||
payload = {
|
||||
"email": email,
|
||||
"exp": int(time.time()) + expires_in,
|
||||
"nonce": secrets.token_hex(16), # Extra randomness
|
||||
"purpose": "password_reset"
|
||||
}
|
||||
|
||||
# Convert to JSON and encode
|
||||
payload_bytes = json.dumps(payload).encode('utf-8')
|
||||
|
||||
# Create a signature using the secret key
|
||||
signature = hashlib.sha256(
|
||||
payload_bytes + settings.SECRET_KEY.encode('utf-8')
|
||||
).hexdigest()
|
||||
|
||||
# Combine payload and signature
|
||||
token_data = {
|
||||
"payload": payload,
|
||||
"signature": signature
|
||||
}
|
||||
|
||||
# Encode the final token
|
||||
token_json = json.dumps(token_data)
|
||||
token = base64.urlsafe_b64encode(token_json.encode('utf-8')).decode('utf-8')
|
||||
|
||||
return token
|
||||
|
||||
|
||||
def verify_password_reset_token(token: str) -> Optional[str]:
|
||||
"""
|
||||
Verify a password reset token and return the email if valid.
|
||||
|
||||
Args:
|
||||
token: The token string to verify
|
||||
|
||||
Returns:
|
||||
The email address if valid, None if invalid or expired
|
||||
|
||||
Example:
|
||||
>>> email = verify_password_reset_token(token_from_user)
|
||||
>>> if email:
|
||||
... # Proceed with password reset
|
||||
... else:
|
||||
... # Token invalid or expired
|
||||
"""
|
||||
try:
|
||||
# Decode the token
|
||||
token_json = base64.urlsafe_b64decode(token.encode('utf-8')).decode('utf-8')
|
||||
token_data = json.loads(token_json)
|
||||
|
||||
# Extract payload and signature
|
||||
payload = token_data["payload"]
|
||||
signature = token_data["signature"]
|
||||
|
||||
# Verify it's a password reset token
|
||||
if payload.get("purpose") != "password_reset":
|
||||
return None
|
||||
|
||||
# Verify signature
|
||||
payload_bytes = json.dumps(payload).encode('utf-8')
|
||||
expected_signature = hashlib.sha256(
|
||||
payload_bytes + settings.SECRET_KEY.encode('utf-8')
|
||||
).hexdigest()
|
||||
|
||||
if signature != expected_signature:
|
||||
return None
|
||||
|
||||
# Check expiration
|
||||
if payload["exp"] < int(time.time()):
|
||||
return None
|
||||
|
||||
return payload["email"]
|
||||
|
||||
except (ValueError, KeyError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def create_email_verification_token(email: str, expires_in: int = 86400) -> str:
|
||||
"""
|
||||
Create a signed token for email verification.
|
||||
|
||||
Args:
|
||||
email: User's email address
|
||||
expires_in: Expiration time in seconds (default: 86400 = 24 hours)
|
||||
|
||||
Returns:
|
||||
A base64 encoded token string
|
||||
|
||||
Example:
|
||||
>>> token = create_email_verification_token("user@example.com")
|
||||
>>> # Send token to user via email
|
||||
"""
|
||||
# Create the payload
|
||||
payload = {
|
||||
"email": email,
|
||||
"exp": int(time.time()) + expires_in,
|
||||
"nonce": secrets.token_hex(16),
|
||||
"purpose": "email_verification"
|
||||
}
|
||||
|
||||
# Convert to JSON and encode
|
||||
payload_bytes = json.dumps(payload).encode('utf-8')
|
||||
|
||||
# Create a signature using the secret key
|
||||
signature = hashlib.sha256(
|
||||
payload_bytes + settings.SECRET_KEY.encode('utf-8')
|
||||
).hexdigest()
|
||||
|
||||
# Combine payload and signature
|
||||
token_data = {
|
||||
"payload": payload,
|
||||
"signature": signature
|
||||
}
|
||||
|
||||
# Encode the final token
|
||||
token_json = json.dumps(token_data)
|
||||
token = base64.urlsafe_b64encode(token_json.encode('utf-8')).decode('utf-8')
|
||||
|
||||
return token
|
||||
|
||||
|
||||
def verify_email_verification_token(token: str) -> Optional[str]:
|
||||
"""
|
||||
Verify an email verification token and return the email if valid.
|
||||
|
||||
Args:
|
||||
token: The token string to verify
|
||||
|
||||
Returns:
|
||||
The email address if valid, None if invalid or expired
|
||||
|
||||
Example:
|
||||
>>> email = verify_email_verification_token(token_from_user)
|
||||
>>> if email:
|
||||
... # Mark email as verified
|
||||
... else:
|
||||
... # Token invalid or expired
|
||||
"""
|
||||
try:
|
||||
# Decode the token
|
||||
token_json = base64.urlsafe_b64decode(token.encode('utf-8')).decode('utf-8')
|
||||
token_data = json.loads(token_json)
|
||||
|
||||
# Extract payload and signature
|
||||
payload = token_data["payload"]
|
||||
signature = token_data["signature"]
|
||||
|
||||
# Verify it's an email verification token
|
||||
if payload.get("purpose") != "email_verification":
|
||||
return None
|
||||
|
||||
# Verify signature
|
||||
payload_bytes = json.dumps(payload).encode('utf-8')
|
||||
expected_signature = hashlib.sha256(
|
||||
payload_bytes + settings.SECRET_KEY.encode('utf-8')
|
||||
).hexdigest()
|
||||
|
||||
if signature != expected_signature:
|
||||
return None
|
||||
|
||||
# Check expiration
|
||||
if payload["exp"] < int(time.time()):
|
||||
return None
|
||||
|
||||
return payload["email"]
|
||||
|
||||
except (ValueError, KeyError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user