from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Any from cryptography.fernet import Fernet from jose import JWTError, jwt from passlib.context import CryptContext from .config import get_settings settings = get_settings() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") ALGORITHM = "HS256" def hash_password(password: str) -> str: return pwd_context.hash(password) def verify_password(plain: str, hashed: str) -> bool: return pwd_context.verify(plain, hashed) def _now() -> datetime: return datetime.now(timezone.utc) def create_access_token(subject: str | int, extra: dict[str, Any] | None = None) -> str: payload: dict[str, Any] = { "sub": str(subject), "type": "access", "iat": _now(), "exp": _now() + timedelta(minutes=settings.access_token_expire_minutes), } if extra: payload.update(extra) return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def create_refresh_token(subject: str | int) -> str: payload = { "sub": str(subject), "type": "refresh", "iat": _now(), "exp": _now() + timedelta(days=settings.refresh_token_expire_days), } return jwt.encode(payload, settings.secret_key, algorithm=ALGORITHM) def decode_token(token: str) -> dict[str, Any]: try: return jwt.decode(token, settings.secret_key, algorithms=[ALGORITHM]) except JWTError as exc: # pragma: no cover raise ValueError(f"invalid token: {exc}") from exc # --- Симметричное шифрование секретов устройств ----------------------------- # Производный ключ из SECRET_KEY (для dev). В prod — KMS / Vault. def _fernet() -> Fernet: import base64 import hashlib digest = hashlib.sha256(settings.secret_key.encode()).digest() key = base64.urlsafe_b64encode(digest) return Fernet(key) def encrypt_secret(value: str) -> str: return _fernet().encrypt(value.encode()).decode() def decrypt_secret(token: str) -> str: return _fernet().decrypt(token.encode()).decode()