Spaces:
Running
Running
| import hashlib | |
| import logging | |
| import secrets | |
| from datetime import datetime, timedelta, timezone | |
| from typing import Optional | |
| from jose import JWTError, jwt | |
| from passlib.context import CryptContext | |
| from app.core.config import settings | |
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") | |
| logger = logging.getLogger(__name__) | |
| def hash_password(password: str) -> str: | |
| return pwd_context.hash(password) | |
| def verify_password(plain: str, hashed: str) -> bool: | |
| return pwd_context.verify(plain, hashed) | |
| def generate_session_id() -> str: | |
| """Unique session ID stored in DB — invalidates old sessions.""" | |
| return secrets.token_hex(32) | |
| def generate_password_reset_token() -> str: | |
| return secrets.token_urlsafe(48) | |
| def hash_password_reset_token(token: str) -> str: | |
| return hashlib.sha256(token.encode("utf-8")).hexdigest() | |
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: | |
| to_encode = data.copy() | |
| expire = datetime.now(timezone.utc) + ( | |
| expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) | |
| ) | |
| to_encode["exp"] = expire | |
| if "sub" in to_encode: | |
| to_encode["sub"] = str(to_encode["sub"]) | |
| return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) | |
| def decode_token(token: str) -> Optional[dict]: | |
| try: | |
| payload = jwt.decode( | |
| token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM] | |
| ) | |
| return payload | |
| except JWTError as e: | |
| logger.warning("JWT decode error: %s", e) | |
| return None | |