Spaces:
Sleeping
Sleeping
| from datetime import datetime, timedelta, timezone | |
| from passlib.context import CryptContext | |
| from jose import JWTError, jwt | |
| from app.core.config import settings | |
| # Password Hashing Context | |
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") | |
| # --- Password Functions --- | |
| 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) | |
| # --- JWT Access Token Functions --- | |
| def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str: | |
| to_encode = data.copy() | |
| if expires_delta: | |
| expire = datetime.now(timezone.utc) + expires_delta | |
| else: | |
| expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) | |
| to_encode.update({"exp": expire, "iat": datetime.now(timezone.utc)}) | |
| encoded_jwt = jwt.encode(to_encode, settings.APP_SECRET_KEY, algorithm=settings.ALGORITHM) | |
| return encoded_jwt | |
| # --- Special Purpose Token Functions (Email Verification / Password Reset) --- | |
| def generate_email_verification_token(email: str) -> str: | |
| """Generates a short-lived JWT for email verification.""" | |
| expires = timedelta(hours=24) | |
| to_encode = { | |
| "exp": datetime.now(timezone.utc) + expires, | |
| "iat": datetime.now(timezone.utc), | |
| "sub": email, | |
| "scope": "email_verification" | |
| } | |
| return jwt.encode(to_encode, settings.APP_SECRET_KEY, algorithm=settings.ALGORITHM) | |
| def verify_email_verification_token(token: str) -> str | None: | |
| """Verifies the email verification token and returns the email.""" | |
| try: | |
| payload = jwt.decode(token, settings.APP_SECRET_KEY, algorithms=[settings.ALGORITHM]) | |
| if payload.get("scope") == "email_verification": | |
| return payload.get("sub") | |
| return None | |
| except JWTError: | |
| return None | |
| def generate_password_reset_token(email: str) -> str: | |
| """ | |
| Generates a short-lived JWT for password reset. | |
| Expires in 15 minutes for enhanced security. | |
| """ | |
| expires = timedelta(minutes=15) | |
| to_encode = { | |
| "exp": datetime.now(timezone.utc) + expires, | |
| "iat": datetime.now(timezone.utc), | |
| "sub": email, | |
| "scope": "password_reset" | |
| } | |
| return jwt.encode(to_encode, settings.APP_SECRET_KEY, algorithm=settings.ALGORITHM) | |
| def verify_password_reset_token(token: str) -> str | None: | |
| """ | |
| Verifies the password reset token and returns the email if valid. | |
| Returns None if token is invalid, expired, or has the wrong scope. | |
| """ | |
| try: | |
| payload = jwt.decode(token, settings.APP_SECRET_KEY, algorithms=[settings.ALGORITHM]) | |
| if payload.get("scope") == "password_reset": | |
| return payload.get("sub") | |
| return None | |
| except JWTError: | |
| # Standard JWT exceptions (ExpiredSignatureError, JWTClaimsError, etc.) are caught here | |
| return None |