| from passlib.context import CryptContext |
| from jose import JWTError, jwt |
| from datetime import datetime, timedelta |
| from typing import Optional |
| from .config import settings |
|
|
| |
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") |
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool: |
| """Verify a plain password against its hash""" |
| return pwd_context.verify(plain_password, hashed_password) |
|
|
| def get_password_hash(password: str) -> str: |
| """Hash a password""" |
| return pwd_context.hash(password) |
|
|
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: |
| """Create a JWT access token""" |
| to_encode = data.copy() |
| if expires_delta: |
| expire = datetime.utcnow() + expires_delta |
| else: |
| expire = datetime.utcnow() + timedelta(minutes=settings.access_token_expire_minutes) |
| |
| to_encode.update({"exp": expire}) |
| encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm) |
| return encoded_jwt |
|
|
| def verify_token(token: str) -> Optional[str]: |
| """Verify JWT token and return email""" |
| try: |
| payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm]) |
| email: str = payload.get("sub") |
| if email is None: |
| return None |
| return email |
| except JWTError: |
| return None |
|
|
