Spaces:
Sleeping
Sleeping
| from datetime import datetime, timedelta | |
| from typing import Optional | |
| import bcrypt | |
| from jose import JWTError, jwt | |
| from src.config import settings | |
| # JWT settings | |
| SECRET_KEY = settings.AUTH_SECRET | |
| ALGORITHM = settings.JWT_ALGORITHM | |
| ACCESS_TOKEN_EXPIRE_MINUTES = settings.JWT_EXPIRATION_MINUTES | |
| def verify_password(plain_password: str, hashed_password: str) -> bool: | |
| """ | |
| Verify a plain password against a hashed password. | |
| Args: | |
| plain_password: The plain text password to verify | |
| hashed_password: The hashed password to compare against | |
| Returns: | |
| bool: True if password matches, False otherwise | |
| """ | |
| # Ensure password is encoded as bytes (bcrypt limitation: max 72 bytes) | |
| password_bytes = plain_password.encode('utf-8') | |
| if len(password_bytes) > 72: | |
| password_bytes = password_bytes[:72] | |
| # Verify the password | |
| try: | |
| return bcrypt.checkpw(password_bytes, hashed_password.encode('utf-8')) | |
| except (ValueError, TypeError): | |
| return False | |
| def get_password_hash(password: str) -> str: | |
| """ | |
| Generate a hash for the given password. | |
| Args: | |
| password: The plain text password to hash | |
| Returns: | |
| str: The hashed password | |
| """ | |
| # Ensure password is encoded as bytes (bcrypt limitation: max 72 bytes) | |
| password_bytes = password.encode('utf-8') | |
| if len(password_bytes) > 72: | |
| password_bytes = password_bytes[:72] | |
| # Generate salt and hash | |
| salt = bcrypt.gensalt() | |
| hashed = bcrypt.hashpw(password_bytes, salt) | |
| return hashed.decode('utf-8') | |
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: | |
| """ | |
| Create a JWT access token. | |
| Args: | |
| data: The data to include in the token (typically user info) | |
| expires_delta: Optional expiration time delta (defaults to 7 days) | |
| Returns: | |
| str: The encoded JWT 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 | |
| def verify_token(token: str) -> Optional[dict]: | |
| """ | |
| Verify a JWT token and return the payload if valid. | |
| Args: | |
| token: The JWT token to verify | |
| Returns: | |
| dict: The token payload if valid, None if invalid | |
| """ | |
| try: | |
| payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) | |
| return payload | |
| except JWTError: | |
| return None |