CRM Bot
102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
"""Password hashing and JWT encoding/decoding.
|
|
|
|
Uses python-jose[cryptography] (per 02-architecture.md Section 13.1) and
|
|
passlib[bcrypt] with bcrypt 4.0.1 (per Section 13.6, 03a-patterns R-6).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from jose import JWTError, jwt
|
|
from passlib.context import CryptContext
|
|
|
|
from app.core.config import get_settings
|
|
|
|
_settings = get_settings()
|
|
|
|
# === Password Hashing ===
|
|
|
|
# CryptContext auto-upgrades old hashes when verified. bcrypt 4.0.1 is pinned
|
|
# because 4.1+ breaks passlib (per 03a-patterns-summary R-6).
|
|
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
|
|
|
|
|
def hash_password(plain: str) -> str:
|
|
"""Hash a plaintext password using bcrypt with the configured rounds."""
|
|
return pwd_context.hash(plain)
|
|
|
|
|
|
def verify_password(plain: str, hashed: str) -> bool:
|
|
"""Verify a plaintext password against a bcrypt hash."""
|
|
try:
|
|
return pwd_context.verify(plain, hashed)
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
# === JWT ===
|
|
|
|
|
|
def create_access_token(
|
|
user_id: int,
|
|
org_id: int,
|
|
role: str,
|
|
expires_delta: timedelta | None = None,
|
|
) -> str:
|
|
"""Create a JWT access token with the given user, org, and role claims.
|
|
|
|
The token contains:
|
|
- sub: user id (string)
|
|
- org_id: organization id
|
|
- role: user role
|
|
- exp: expiry timestamp
|
|
- iat: issued-at timestamp
|
|
"""
|
|
now = datetime.now(UTC)
|
|
if expires_delta is None:
|
|
expires_delta = timedelta(hours=_settings.JWT_EXPIRY_HOURS)
|
|
expire = now + expires_delta
|
|
|
|
payload: dict[str, Any] = {
|
|
"sub": str(user_id),
|
|
"org_id": org_id,
|
|
"role": role,
|
|
"iat": int(now.timestamp()),
|
|
"exp": int(expire.timestamp()),
|
|
}
|
|
return jwt.encode(payload, _settings.AUTH_SECRET, algorithm=_settings.JWT_ALGORITHM)
|
|
|
|
|
|
def decode_access_token(token: str) -> dict[str, Any] | None:
|
|
"""Decode and validate a JWT access token.
|
|
|
|
Returns the payload dict on success, or None on any error
|
|
(invalid signature, malformed token, expired).
|
|
"""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
_settings.AUTH_SECRET,
|
|
algorithms=[_settings.JWT_ALGORITHM],
|
|
)
|
|
return payload
|
|
except JWTError:
|
|
return None
|
|
|
|
|
|
def is_token_expired(token: str) -> bool:
|
|
"""Check whether a token is expired without raising on other errors."""
|
|
try:
|
|
jwt.get_unverified_claims(token)
|
|
# If decode succeeds, it's not expired.
|
|
jwt.decode(
|
|
token, _settings.AUTH_SECRET, algorithms=[_settings.JWT_ALGORITHM]
|
|
)
|
|
return False
|
|
except JWTError as e:
|
|
return "expired" in str(e).lower() or "exp" in str(e).lower()
|
|
except Exception:
|
|
return True
|