| from __future__ import annotations |
|
|
| from datetime import UTC, datetime, timedelta |
|
|
| import bcrypt |
| import jwt |
|
|
| from app.config import settings |
|
|
|
|
| def hash_password(password: str) -> str: |
| salt = bcrypt.gensalt() |
| return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8") |
|
|
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool: |
| try: |
| return bcrypt.checkpw( |
| plain_password.encode("utf-8"), |
| hashed_password.encode("utf-8"), |
| ) |
| except ValueError: |
| return False |
|
|
|
|
| def create_access_token(subject: str, expires_delta: timedelta | None = None) -> str: |
| expire = datetime.now(UTC) + ( |
| expires_delta |
| if expires_delta is not None |
| else timedelta(minutes=settings.access_token_expire_minutes) |
| ) |
| payload = {"sub": subject, "exp": expire} |
| return jwt.encode(payload, settings.jwt_secret_key, algorithm=settings.jwt_algorithm) |
|
|
|
|
| def decode_access_token(token: str) -> dict: |
| return jwt.decode( |
| token, |
| settings.jwt_secret_key, |
| algorithms=[settings.jwt_algorithm], |
| ) |
|
|