120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
"""Users helpers — password hashing, JWT tokens, and payload cleaning.
|
|
|
|
Uses the `bcrypt` package directly rather than passlib: passlib 1.7.4 reads
|
|
`bcrypt.__about__.__version__`, which bcrypt dropped in 4.1, and the failed
|
|
version probe makes it reject every password as longer than 72 bytes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import bcrypt
|
|
import jwt
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# bcrypt hashes at most 72 bytes and raises on anything longer.
|
|
BCRYPT_MAX_BYTES = 72
|
|
|
|
# Columns the server owns; a client must never be able to set them.
|
|
SERVER_OWNED_FIELDS = ("id", "created_at", "updated_at", "is_deleted")
|
|
|
|
JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY")
|
|
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
|
|
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("JWT_ACCESS_TOKEN_EXPIRE_MINUTES", "30"))
|
|
REFRESH_TOKEN_EXPIRE_DAYS = int(os.getenv("JWT_REFRESH_TOKEN_EXPIRE_DAYS", "7"))
|
|
ACCESS_TOKEN_EXPIRE_SECONDS = ACCESS_TOKEN_EXPIRE_MINUTES * 60
|
|
|
|
|
|
def _encode(raw: str) -> bytes:
|
|
"""UTF-8 bytes truncated to what bcrypt accepts, without splitting a character."""
|
|
return raw.encode("utf-8")[:BCRYPT_MAX_BYTES].decode("utf-8", "ignore").encode("utf-8")
|
|
|
|
|
|
def hash_password(raw: str) -> str:
|
|
return bcrypt.hashpw(_encode(raw), bcrypt.gensalt()).decode("ascii")
|
|
|
|
|
|
def verify_password(raw: str, hashed: str) -> bool:
|
|
"""False rather than raising on rows written before hashing existed."""
|
|
if not raw or not hashed:
|
|
return False
|
|
try:
|
|
return bcrypt.checkpw(_encode(raw), hashed.encode("utf-8"))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def clean_user_payload(payload: dict, *, partial: bool = False) -> dict:
|
|
"""Strip server-owned keys and hash the password; on partial, drop unset fields."""
|
|
fields = {
|
|
key: value
|
|
for key, value in payload.items()
|
|
if key not in SERVER_OWNED_FIELDS
|
|
}
|
|
if partial:
|
|
fields = {key: value for key, value in fields.items() if value is not None}
|
|
if fields.get("password"):
|
|
fields["password"] = hash_password(fields["password"])
|
|
else:
|
|
fields.pop("password", None)
|
|
return fields
|
|
|
|
|
|
def _secret() -> str:
|
|
if not JWT_SECRET_KEY:
|
|
raise RuntimeError("JWT_SECRET_KEY is not set")
|
|
return JWT_SECRET_KEY
|
|
|
|
|
|
def _create_token(
|
|
subject: str,
|
|
*,
|
|
token_type: str,
|
|
expires_delta: timedelta,
|
|
claims: dict[str, Any] | None = None,
|
|
) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
payload: dict[str, Any] = {
|
|
"sub": subject,
|
|
"type": token_type,
|
|
"iat": now,
|
|
"exp": now + expires_delta,
|
|
"jti": str(uuid.uuid4()),
|
|
}
|
|
if claims:
|
|
payload.update(claims)
|
|
return jwt.encode(payload, _secret(), algorithm=JWT_ALGORITHM)
|
|
|
|
|
|
def create_access_token(user) -> str:
|
|
return _create_token(
|
|
str(user.id),
|
|
token_type="access",
|
|
expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
|
|
claims={
|
|
"email": user.email,
|
|
"role_id": user.role_id,
|
|
},
|
|
)
|
|
|
|
|
|
def create_refresh_token(user) -> str:
|
|
return _create_token(
|
|
str(user.id),
|
|
token_type="refresh",
|
|
expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS),
|
|
)
|
|
|
|
|
|
def decode_token(token: str, *, expected_type: str) -> dict:
|
|
payload = jwt.decode(token, _secret(), algorithms=[JWT_ALGORITHM])
|
|
if payload.get("type") != expected_type:
|
|
raise jwt.InvalidTokenError("Unexpected token type")
|
|
return payload
|