Spaces:
Runtime error
Runtime error
File size: 5,102 Bytes
aad7814 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """Auth primitives: password hashing, JWT, and a small file-based user store.
Prefers ``passlib`` / ``python-jose`` when installed (see requirements.txt) and
falls back to stdlib implementations (PBKDF2-HMAC + a hand-rolled HS256 token) so
the backend runs and tests pass without those optional dependencies.
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import os
import time
from pathlib import Path
from backend.config import settings
# ── password hashing ─────────────────────────────────────────────────────────
try:
import bcrypt
def hash_password(password: str) -> str:
return bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt()).decode("ascii")
def verify_password(password: str, hashed: str) -> bool:
try:
return bcrypt.checkpw(password.encode("utf-8"), hashed.encode("ascii"))
except Exception: # noqa: BLE001
return False
except ImportError: # bcrypt unavailable → PBKDF2 fallback
def hash_password(password: str) -> str:
salt = os.urandom(16)
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, 120_000)
return f"pbkdf2_sha256$120000${salt.hex()}${dk.hex()}"
def verify_password(password: str, hashed: str) -> bool:
try:
algo, iters, salt_hex, dk_hex = hashed.split("$")
except ValueError:
return False
if algo != "pbkdf2_sha256":
return False
dk = hashlib.pbkdf2_hmac(
"sha256", password.encode("utf-8"), bytes.fromhex(salt_hex), int(iters)
)
return hmac.compare_digest(dk.hex(), dk_hex)
# ── JWT ──────────────────────────────────────────────────────────────────────
def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
def _b64url_decode(data: str) -> bytes:
pad = "=" * (-len(data) % 4)
return base64.urlsafe_b64decode(data + pad)
try:
from jose import jwt as _jose_jwt
def create_token(tenant_id: str) -> str:
payload = {
"sub": tenant_id,
"exp": int(time.time()) + settings.jwt_expiry_minutes * 60,
}
return _jose_jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def decode_token(token: str) -> dict:
return _jose_jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
except Exception: # noqa: BLE001 - python-jose unavailable → HS256 fallback
def create_token(tenant_id: str) -> str:
header = _b64url(json.dumps({"alg": "HS256", "typ": "JWT"}).encode())
payload = _b64url(json.dumps({
"sub": tenant_id,
"exp": int(time.time()) + settings.jwt_expiry_minutes * 60,
}).encode())
signing_input = f"{header}.{payload}".encode()
sig = hmac.new(settings.jwt_secret.encode(), signing_input, hashlib.sha256).digest()
return f"{header}.{payload}.{_b64url(sig)}"
def decode_token(token: str) -> dict:
try:
header_b64, payload_b64, sig_b64 = token.split(".")
except ValueError as exc:
raise ValueError("Malformed token") from exc
signing_input = f"{header_b64}.{payload_b64}".encode()
expected = hmac.new(settings.jwt_secret.encode(), signing_input, hashlib.sha256).digest()
if not hmac.compare_digest(expected, _b64url_decode(sig_b64)):
raise ValueError("Invalid signature")
payload = json.loads(_b64url_decode(payload_b64))
if int(payload.get("exp", 0)) < int(time.time()):
raise ValueError("Token expired")
return payload
# ── user store ───────────────────────────────────────────────────────────────
def _users_path() -> Path:
p = settings.data_dir_path / "users.json"
p.parent.mkdir(parents=True, exist_ok=True)
return p
def _load_users() -> dict[str, str]:
path = _users_path()
if not path.is_file():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001
return {}
def _save_users(users: dict[str, str]) -> None:
_users_path().write_text(json.dumps(users), encoding="utf-8")
def register_user(tenant_id: str, password: str) -> bool:
"""Create a tenant. Returns False if it already exists."""
users = _load_users()
if tenant_id in users:
return False
users[tenant_id] = hash_password(password)
_save_users(users)
return True
def authenticate(tenant_id: str, password: str) -> bool:
users = _load_users()
hashed = users.get(tenant_id)
return bool(hashed and verify_password(password, hashed))
|