54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
|
from datetime import datetime, timedelta, timezone
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from jose import JWTError, jwt
|
||
|
|
from passlib.context import CryptContext
|
||
|
|
|
||
|
|
from app.core.config import settings
|
||
|
|
|
||
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||
|
|
|
||
|
|
ALGORITHM = "HS256"
|
||
|
|
TOKEN_TYPE_ACCESS = "access"
|
||
|
|
TOKEN_TYPE_REFRESH = "refresh"
|
||
|
|
|
||
|
|
|
||
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
||
|
|
return pwd_context.verify(plain_password, hashed_password)
|
||
|
|
|
||
|
|
|
||
|
|
def get_password_hash(password: str) -> str:
|
||
|
|
return pwd_context.hash(password)
|
||
|
|
|
||
|
|
|
||
|
|
def create_access_token(subject: str | Any, extra_claims: dict | None = None) -> str:
|
||
|
|
expire = datetime.now(timezone.utc) + timedelta(
|
||
|
|
minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES
|
||
|
|
)
|
||
|
|
to_encode: dict[str, Any] = {
|
||
|
|
"sub": str(subject),
|
||
|
|
"exp": expire,
|
||
|
|
"type": TOKEN_TYPE_ACCESS,
|
||
|
|
}
|
||
|
|
if extra_claims:
|
||
|
|
to_encode.update(extra_claims)
|
||
|
|
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
|
||
|
|
|
||
|
|
|
||
|
|
def create_refresh_token(subject: str | Any) -> str:
|
||
|
|
expire = datetime.now(timezone.utc) + timedelta(
|
||
|
|
days=settings.REFRESH_TOKEN_EXPIRE_DAYS
|
||
|
|
)
|
||
|
|
to_encode: dict[str, Any] = {
|
||
|
|
"sub": str(subject),
|
||
|
|
"exp": expire,
|
||
|
|
"type": TOKEN_TYPE_REFRESH,
|
||
|
|
}
|
||
|
|
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=ALGORITHM)
|
||
|
|
|
||
|
|
|
||
|
|
def decode_token(token: str) -> dict[str, Any]:
|
||
|
|
"""Decode and validate a JWT. Raises JWTError on failure."""
|
||
|
|
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[ALGORITHM])
|
||
|
|
return payload
|