| """ |
| JWT Security, Password Hashing, and RBAC utilities. |
| """ |
| from __future__ import annotations |
|
|
| from datetime import datetime, timedelta |
| from enum import Enum |
| from typing import Optional |
|
|
| import structlog |
| from jose import JWTError, jwt |
| from passlib.context import CryptContext |
|
|
| from app.core.config import settings |
|
|
| logger = structlog.get_logger(__name__) |
|
|
| import bcrypt |
| import passlib.handlers.bcrypt |
|
|
| def _fixed_calc_checksum(self, secret): |
| if isinstance(secret, str): |
| secret = secret.encode("utf-8") |
| secret = secret[:72] |
| secret, ident = self._prepare_digest_args(secret) |
| config = self._get_config(ident) |
| if isinstance(config, str): |
| config = config.encode("ascii") |
| raw_hash = bcrypt.hashpw(secret, config).decode("utf-8") |
| return raw_hash[29:] |
|
|
| passlib.handlers.bcrypt._BcryptBackend._calc_checksum = _fixed_calc_checksum |
| passlib.handlers.bcrypt._BcryptBackend._lacks_20_support = True |
| passlib.handlers.bcrypt._BcryptBackend._workrounds_initialized = True |
|
|
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") |
|
|
|
|
| |
| class UserRole(str, Enum): |
| ADMIN = "admin" |
| HR = "hr" |
| FINANCE = "finance" |
| EMPLOYEE = "employee" |
|
|
|
|
| |
| ROLE_PERMISSIONS: dict[UserRole, list[str]] = { |
| UserRole.ADMIN: [ |
| "chat:all", "documents:all", "sql:all", |
| "admin:all", "reports:all", "users:manage", |
| ], |
| UserRole.FINANCE: [ |
| "chat:basic", "sql:finance", "documents:finance", |
| "reports:finance", "analytics:finance", |
| ], |
| UserRole.HR: [ |
| "chat:basic", "documents:hr", "reports:hr", |
| "analytics:hr", |
| ], |
| UserRole.EMPLOYEE: [ |
| "chat:basic", "documents:public", |
| ], |
| } |
|
|
|
|
| |
| def hash_password(password: str) -> str: |
| pwd_clean = password.encode("utf-8")[:72].decode("utf-8", errors="ignore") |
| return pwd_context.hash(pwd_clean) |
|
|
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool: |
| pwd_clean = plain_password.encode("utf-8")[:72].decode("utf-8", errors="ignore") |
| return pwd_context.verify(pwd_clean, hashed_password) |
|
|
|
|
| |
| def create_access_token( |
| user_id: str, |
| tenant_id: str, |
| role: UserRole, |
| email: str, |
| expires_delta: Optional[timedelta] = None, |
| ) -> str: |
| expire = datetime.utcnow() + ( |
| expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) |
| ) |
| payload = { |
| "sub": user_id, |
| "tenant_id": tenant_id, |
| "role": role.value, |
| "email": email, |
| "type": "access", |
| "exp": expire, |
| } |
| return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM) |
|
|
|
|
| def create_refresh_token(user_id: str, tenant_id: str) -> str: |
| expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS) |
| payload = { |
| "sub": user_id, |
| "tenant_id": tenant_id, |
| "type": "refresh", |
| "exp": expire, |
| } |
| return jwt.encode(payload, settings.SECRET_KEY, algorithm=settings.JWT_ALGORITHM) |
|
|
|
|
| def decode_token(token: str) -> dict: |
| """Decode and validate a JWT token. Raises JWTError on failure.""" |
| return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.JWT_ALGORITHM]) |
|
|
|
|
| |
| def has_permission(role: UserRole, permission: str) -> bool: |
| """Check if a role has a specific permission.""" |
| role_perms = ROLE_PERMISSIONS.get(role, []) |
| |
| return ( |
| permission in role_perms |
| or any( |
| perm.endswith(":all") and perm.split(":")[0] == permission.split(":")[0] |
| for perm in role_perms |
| ) |
| or "admin:all" in role_perms |
| ) |
|
|
|
|
| def require_roles(*roles: UserRole): |
| """FastAPI dependency factory: require specific roles.""" |
| from typing import Any |
| from fastapi import Depends, HTTPException, status |
| from app.core.dependencies import get_current_active_user |
|
|
| async def role_checker(current_user: Any = Depends(get_current_active_user)): |
| if current_user.role not in roles: |
| raise HTTPException( |
| status_code=status.HTTP_403_FORBIDDEN, |
| detail={ |
| "error": "insufficient_permissions", |
| "message": f"Required role(s): {[r.value for r in roles]}", |
| "your_role": current_user.role, |
| }, |
| ) |
| return current_user |
|
|
| return role_checker |
|
|
