"""JWT creation/validation and password hashing.""" from __future__ import annotations import logging from datetime import datetime, timedelta, timezone import bcrypt from jose import JWTError, jwt logger = logging.getLogger(__name__) def hash_password(password: str) -> str: """Hash a password using bcrypt. Returns UTF-8 encoded hash string.""" password_bytes = password.encode("utf-8") salt = bcrypt.gensalt() hashed = bcrypt.hashpw(password_bytes, salt) return hashed.decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: """Verify a plain password against a bcrypt hash.""" try: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) except Exception as exc: logger.warning("Password verification failed: %s", exc) return False def create_access_token(user_id: int, username: str, role: str) -> str: from config import settings expire = datetime.now(timezone.utc) + timedelta(hours=settings.jwt_expire_hours) payload = { "sub": str(user_id), "username": username, "role": role, "exp": expire, } return jwt.encode(payload, settings.secret_key, algorithm="HS256") def decode_access_token(token: str) -> dict: """ Decode and validate JWT. Returns payload dict. Raises JWTError on invalid/expired token. """ from config import settings return jwt.decode(token, settings.secret_key, algorithms=["HS256"])