| from datetime import datetime, timedelta, timezone |
| from typing import Any |
|
|
| import bcrypt |
| import jwt |
|
|
| from core.config import settings |
|
|
|
|
| _BCRYPT_ROUNDS = 12 |
| _BCRYPT_MAX_BYTES = 72 |
|
|
|
|
| def _prepare_password(password: str) -> bytes: |
| encoded = password.encode("utf-8") |
| if len(encoded) > _BCRYPT_MAX_BYTES: |
| encoded = encoded[:_BCRYPT_MAX_BYTES] |
| return encoded |
|
|
|
|
| def hash_password(password: str) -> str: |
| return bcrypt.hashpw(_prepare_password(password), bcrypt.gensalt(rounds=_BCRYPT_ROUNDS)).decode("utf-8") |
|
|
|
|
| def verify_password(password: str, hashed: str) -> bool: |
| try: |
| return bcrypt.checkpw(_prepare_password(password), hashed.encode("utf-8")) |
| except ValueError: |
| return False |
|
|
|
|
| def create_access_token(subject: str | int, extra_claims: dict[str, Any] | None = None) -> tuple[str, int]: |
| now = datetime.now(timezone.utc) |
| expires_at = now + timedelta(minutes=settings.JWT_EXPIRE_MINUTES) |
| payload: dict[str, Any] = { |
| "sub": str(subject), |
| "iat": int(now.timestamp()), |
| "exp": int(expires_at.timestamp()), |
| "type": "access", |
| } |
| if extra_claims: |
| payload.update(extra_claims) |
| token = jwt.encode(payload, settings.JWT_SECRET, algorithm=settings.JWT_ALGORITHM) |
| return token, settings.JWT_EXPIRE_MINUTES * 60 |
|
|
|
|
| def decode_token(token: str) -> dict[str, Any]: |
| return jwt.decode(token, settings.JWT_SECRET, algorithms=[settings.JWT_ALGORITHM]) |
|
|
