40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
import bcrypt
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from fastapi import Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from db.session import get_db
|
|
from models.users import User
|
|
from jose import jwt
|
|
from services.users import get_user_by_id
|
|
from core.settings import JWT_SECRET
|
|
from jose.exceptions import JWTError
|
|
|
|
|
|
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed: str) -> bool:
|
|
return bcrypt.checkpw(
|
|
plain_password.encode("utf-8"),
|
|
hashed.encode("utf-8"),
|
|
)
|
|
|
|
|
|
async def get_current_user(token: str = Depends(oauth2_scheme), db: AsyncSession = Depends(get_db)) -> User:
|
|
try:
|
|
payload = jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
|
|
|
|
id = int(payload["sub"])
|
|
|
|
user = await get_user_by_id(id, db)
|
|
|
|
if not user:
|
|
raise HTTPException(401, "Unauthorized")
|
|
|
|
return user
|
|
except JWTError:
|
|
raise HTTPException(401, "Unauthorized")
|