Edwin Eames
6c35393f1f
Updates the user registration and new account creation endpoints to require email confirmation. - Sets the 'confirmed' flag to 'false' by default for all new user accounts. - Generates a unique confirmation token for each new user. - Logs the confirmation link to the console for development purposes. This change ensures that users cannot log in without first verifying their email address, enhancing account security.
67 lines
2.5 KiB
Python
67 lines
2.5 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from database import get_db
|
|
from models import Account_User
|
|
from schemas import UserLogin, Token
|
|
from passlib.context import CryptContext
|
|
from jose import jwt
|
|
from datetime import datetime, timedelta
|
|
from config import load_config
|
|
|
|
pwd_context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
# Load JWT configuration from environment
|
|
ApplicationConfig = load_config()
|
|
SECRET_KEY = ApplicationConfig.JWT_SECRET_KEY
|
|
ALGORITHM = ApplicationConfig.JWT_ALGORITHM
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = ApplicationConfig.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
|
|
|
|
def verify_password(plain_password, hashed_password):
|
|
return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
def create_access_token(data: dict, expires_delta = None):
|
|
to_encode = data.copy()
|
|
if expires_delta:
|
|
expire = datetime.utcnow() + expires_delta
|
|
else:
|
|
expire = datetime.utcnow() + timedelta(minutes=15)
|
|
to_encode.update({"exp": expire})
|
|
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
return encoded_jwt
|
|
|
|
async def authenticate_user(db: AsyncSession, email: str, password: str):
|
|
user = await db.execute(select(Account_User).where(Account_User.email == email))
|
|
user = user.scalar_one_or_none()
|
|
if not user:
|
|
return False
|
|
if not verify_password(password, user.password_hash):
|
|
return False
|
|
if not user.confirmed:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Please confirm your email address before logging in.",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
# Update last_seen
|
|
user.last_seen = datetime.utcnow()
|
|
await db.commit()
|
|
return user
|
|
|
|
router = APIRouter()
|
|
|
|
@router.post("/login", response_model=Token)
|
|
async def login(user: UserLogin, db: AsyncSession = Depends(get_db)):
|
|
user_obj = await authenticate_user(db, user.email, user.password)
|
|
if not user_obj:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Incorrect email or password",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
access_token = create_access_token(
|
|
data={"sub": user_obj.username}, expires_delta=access_token_expires
|
|
)
|
|
return {"access_token": access_token, "token_type": "bearer"}
|