from datetime import datetime, timedelta import hashlib import hmac from jose import jwt, JWTError from passlib.context import CryptContext from app.core.config import settings pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def create_access_token(subject: str, expires_minutes: int | None = None) -> str: expire = datetime.utcnow() + timedelta( minutes=expires_minutes or settings.access_token_expire_minutes ) to_encode = {"sub": subject, "exp": expire} return jwt.encode( to_encode, settings.jwt_secret_key, algorithm=settings.jwt_algorithm, ) def decode_token(token: str) -> dict: payload = jwt.decode( token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm], ) if payload.get("type", "access") != "access": raise JWTError("Not an access token") return payload def password_stamp(hashed_password: str) -> str: return hmac.new(settings.jwt_secret_key.encode(), hashed_password.encode(), hashlib.sha256).hexdigest() def create_refresh_token(subject: str, hashed_password: str) -> str: return jwt.encode( {"sub": subject, "type": "refresh", "pwd": password_stamp(hashed_password), "exp": datetime.utcnow() + timedelta(days=90)}, settings.jwt_secret_key, algorithm=settings.jwt_algorithm, ) def decode_refresh_token(token: str) -> dict: payload = jwt.decode(token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm]) if payload.get("type") != "refresh": raise JWTError("Not a refresh token") return payload