| 1 | from __future__ import annotations |
| 2 | |
| 3 | import base64 |
| 4 | import hashlib |
| 5 | from datetime import datetime, timedelta, timezone |
| 6 | from typing import Any |
| 7 | |
| 8 | import bcrypt |
| 9 | from jose import jwt |
| 10 | |
| 11 | from app.core.config import settings |
| 12 | |
| 13 | |
| 14 | def _prehash(plain: str) -> bytes: |
| 15 | """bcrypt silently truncates at 72 bytes (and bcrypt>=4.1 raises). Pre-hash with SHA-256 |
| 16 | and base64-encode so passwords of any length are fully mixed into a fixed 44-byte input.""" |
| 17 | digest = hashlib.sha256(plain.encode("utf-8")).digest() |
| 18 | return base64.b64encode(digest) |
| 19 | |
| 20 | |
| 21 | def hash_password(plain: str) -> str: |
| 22 | return bcrypt.hashpw(_prehash(plain), bcrypt.gensalt()).decode("utf-8") |
| 23 | |
| 24 | |
| 25 | def verify_password(plain: str, hashed: str) -> bool: |
| 26 | try: |
| 27 | return bcrypt.checkpw(_prehash(plain), hashed.encode("utf-8")) |
| 28 | except (ValueError, TypeError): |
| 29 | return False |
| 30 | |
| 31 | |
| 32 | def create_access_token(subject: str, role: str, extra: dict[str, Any] | None = None) -> str: |
| 33 | now = datetime.now(timezone.utc) |
| 34 | payload: dict[str, Any] = { |
| 35 | "sub": subject, |
| 36 | "role": role, |
| 37 | "iat": now, |
| 38 | "exp": now + timedelta(minutes=settings.access_token_expire_minutes), |
| 39 | } |
| 40 | if extra: |
| 41 | payload.update(extra) |
| 42 | return jwt.encode(payload, settings.secret_key, algorithm=settings.jwt_algorithm) |
| 43 | |
| 44 | |
| 45 | def decode_token(token: str) -> dict[str, Any]: |
| 46 | return jwt.decode(token, settings.secret_key, algorithms=[settings.jwt_algorithm]) |
| 47 | |
| 48 | |
| 49 | def generate_api_key() -> tuple[str, str]: |
| 50 | """Return (plaintext, sha256). Plaintext is shown to the user once; only the hash is stored. |
| 51 | Prefix `sk_` makes the key recognizable in logs/leak scanners.""" |
| 52 | import secrets |
| 53 | raw = "sk_" + secrets.token_urlsafe(32) |
| 54 | return raw, hash_api_key(raw) |
| 55 | |
| 56 | |
| 57 | def hash_api_key(raw: str) -> str: |
| 58 | return hashlib.sha256(raw.encode("utf-8")).hexdigest() |