Spaces:
Running
Running
| from datetime import datetime, timedelta | |
| from typing import Optional | |
| import os | |
| import bcrypt | |
| from jose import JWTError, jwt | |
| from passlib.context import CryptContext | |
| from dotenv import load_dotenv | |
| # --- BCrypt Compatibility Fix --- | |
| # Satisfy passlib's check for the bcrypt library version (needed for modern Python/BCrypt compatibility) | |
| if not hasattr(bcrypt, "__about__"): | |
| bcrypt.__about__ = type('About', (object,), {'__version__': bcrypt.__version__}) | |
| load_dotenv() | |
| # --- Configuration --- | |
| # Use environment variables or local fallback for development | |
| # SECRET_KEY is used for signing JWT tokens and should be 32+ characters | |
| SECRET_KEY = os.getenv("SECRET_KEY", "f1fda542576b459ee24bf2d459711a439576da2d804dcbb35de9e18261") | |
| ALGORITHM = os.getenv("ALGORITHM", "HS256") | |
| ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", 30)) | |
| # --- Password Management --- | |
| # Using the standard bcrypt hashing scheme with passlib | |
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") | |
| def verify_password(plain_password: str, hashed_password: str) -> bool: | |
| """Verifies a plain text password against its hashed version using bcrypt.""" | |
| try: | |
| if isinstance(plain_password, str): | |
| plain_password = plain_password.encode('utf-8') | |
| if isinstance(hashed_password, str): | |
| hashed_password = hashed_password.encode('utf-8') | |
| return bcrypt.checkpw(plain_password, hashed_password) | |
| except Exception as e: | |
| print(f"Bcrypt verification error: {e}") | |
| return False | |
| def get_password_hash(password: str) -> str: | |
| """Creates a secure bcrypt hash for a password.""" | |
| if isinstance(password, str): | |
| password = password.encode('utf-8') | |
| # Generate salt and hash | |
| salt = bcrypt.gensalt() | |
| hashed = bcrypt.hashpw(password, salt) | |
| return hashed.decode('utf-8') | |
| # --- Token Management --- | |
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: | |
| """Generates a signed JWT access token.""" | |
| to_encode = data.copy() | |
| if expires_delta: | |
| expire = datetime.utcnow() + expires_delta | |
| else: | |
| expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) | |
| to_encode.update({"exp": expire}) | |
| encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) | |
| return encoded_jwt | |