61 lines
2.2 KiB
Python
61 lines
2.2 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
|
|
# 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"}
|