44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
import jwt
|
|
from argon2 import PasswordHasher
|
|
from argon2.exceptions import VerifyMismatchError
|
|
from fastapi import Depends, HTTPException
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from sqlalchemy.orm import Session
|
|
from .config import get_settings
|
|
from .db import get_db
|
|
from .models import User
|
|
|
|
hasher = PasswordHasher()
|
|
oauth2 = OAuth2PasswordBearer(tokenUrl="/auth/token")
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
return hasher.hash(password)
|
|
|
|
|
|
def verify_password(password: str, digest: str) -> bool:
|
|
try:
|
|
return hasher.verify(digest, password)
|
|
except VerifyMismatchError:
|
|
return False
|
|
|
|
|
|
def make_token(user_id: str) -> str:
|
|
cfg = get_settings()
|
|
now = datetime.now(timezone.utc)
|
|
return jwt.encode({"sub": user_id, "iat": now, "exp": now + timedelta(minutes=cfg.access_token_minutes)}, cfg.jwt_secret, algorithm=cfg.jwt_algorithm)
|
|
|
|
|
|
def current_user(token: str = Depends(oauth2), db: Session = Depends(get_db)) -> User:
|
|
cfg = get_settings()
|
|
try:
|
|
user_id = jwt.decode(token, cfg.jwt_secret, algorithms=[cfg.jwt_algorithm])["sub"]
|
|
except (jwt.PyJWTError, KeyError):
|
|
raise HTTPException(401, "Invalid or expired token", headers={"WWW-Authenticate": "Bearer"})
|
|
user = db.get(User, user_id)
|
|
if not user:
|
|
raise HTTPException(401, "Account no longer exists")
|
|
return user
|
|
|