Antoni Nuñez Romeu
127 lines
4.6 KiB
Python
127 lines
4.6 KiB
Python
import hashlib
|
|
import secrets
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
from uuid import UUID, uuid4
|
|
|
|
from jose import JWTError, jwt
|
|
|
|
from src.core.exceptions import ForbiddenException, UnauthorizedException
|
|
from src.domain.entities import UserRole
|
|
from src.infrastructure.config.settings import settings
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
import bcrypt as _bcrypt
|
|
return _bcrypt.hashpw(password.encode("utf-8"), _bcrypt.gensalt()).decode("utf-8")
|
|
|
|
|
|
def verify_password(plain_password: str, hashed_password: str) -> bool:
|
|
import bcrypt as _bcrypt
|
|
return _bcrypt.checkpw(plain_password.encode("utf-8"), hashed_password.encode("utf-8"))
|
|
|
|
|
|
def create_access_token(subject: str | UUID, role: UserRole, scopes: list[str], extra_claims: dict | None = None) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
expires = now + timedelta(minutes=settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
claims: dict[str, Any] = {
|
|
"sub": str(subject),
|
|
"role": role.value,
|
|
"scopes": scopes,
|
|
"iss": settings.JWT_ISSUER,
|
|
"iat": now,
|
|
"exp": expires,
|
|
"type": "access",
|
|
"jti": str(uuid4()),
|
|
}
|
|
if extra_claims:
|
|
claims.update(extra_claims)
|
|
return jwt.encode(claims, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def create_refresh_token(subject: str | UUID) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
expires = now + timedelta(days=settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS)
|
|
claims: dict[str, Any] = {
|
|
"sub": str(subject),
|
|
"iss": settings.JWT_ISSUER,
|
|
"iat": now,
|
|
"exp": expires,
|
|
"type": "refresh",
|
|
"jti": str(uuid4()),
|
|
}
|
|
return jwt.encode(claims, settings.JWT_SECRET_KEY, algorithm=settings.JWT_ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict[str, Any]:
|
|
try:
|
|
payload = jwt.decode(token, settings.JWT_SECRET_KEY, algorithms=[settings.JWT_ALGORITHM], issuer=settings.JWT_ISSUER)
|
|
return payload
|
|
except JWTError as e:
|
|
raise UnauthorizedException(message=f"Invalid token: {e}") from e
|
|
|
|
|
|
def validate_token_type(payload: dict[str, Any], expected_type: str = "access") -> dict[str, Any]:
|
|
if payload.get("type") != expected_type:
|
|
raise UnauthorizedException(message=f"Expected {expected_type} token, got {payload.get('type')}")
|
|
return payload
|
|
|
|
|
|
class TokenData:
|
|
def __init__(self, payload: dict[str, Any]):
|
|
self.subject: str = payload.get("sub", "")
|
|
self.role: UserRole = UserRole(payload.get("role", "viewer"))
|
|
self.scopes: list[str] = payload.get("scopes", [])
|
|
self.issuer: str = payload.get("iss", "")
|
|
self.token_type: str = payload.get("type", "access")
|
|
self.jti: str = payload.get("jti", "")
|
|
self.exp: datetime | None = None
|
|
if exp := payload.get("exp"):
|
|
self.exp = datetime.fromtimestamp(exp, tz=timezone.utc)
|
|
|
|
|
|
class RBACGuard:
|
|
ROLE_HIERARCHY: dict[UserRole, set[UserRole]] = {
|
|
UserRole.SUPERADMIN: {UserRole.SUPERADMIN, UserRole.ADMIN, UserRole.OPERATOR, UserRole.VIEWER, UserRole.API_CLIENT},
|
|
UserRole.ADMIN: {UserRole.ADMIN, UserRole.OPERATOR, UserRole.VIEWER, UserRole.API_CLIENT},
|
|
UserRole.OPERATOR: {UserRole.OPERATOR, UserRole.VIEWER},
|
|
UserRole.VIEWER: {UserRole.VIEWER},
|
|
UserRole.API_CLIENT: {UserRole.API_CLIENT, UserRole.VIEWER},
|
|
}
|
|
|
|
@classmethod
|
|
def has_role(cls, token_data: TokenData, required_role: UserRole) -> bool:
|
|
allowed = cls.ROLE_HIERARCHY.get(token_data.role, set())
|
|
return required_role in allowed
|
|
|
|
@classmethod
|
|
def require_role(cls, token_data: TokenData, required_role: UserRole) -> None:
|
|
if not cls.has_role(token_data, required_role):
|
|
raise ForbiddenException(required_role=required_role.value)
|
|
|
|
@classmethod
|
|
def has_scope(cls, token_data: TokenData, required_scope: str) -> bool:
|
|
if "*" in token_data.scopes:
|
|
return True
|
|
return required_scope in token_data.scopes
|
|
|
|
@classmethod
|
|
def require_scope(cls, token_data: TokenData, required_scope: str) -> None:
|
|
if not cls.has_scope(token_data, required_scope):
|
|
raise ForbiddenException(message=f"Scope '{required_scope}' required")
|
|
|
|
|
|
def generate_api_key() -> tuple[str, str, str]:
|
|
raw_key = secrets.token_urlsafe(settings.API_KEY_LENGTH)
|
|
prefix = raw_key[:8]
|
|
key_hash = hashlib.sha256(raw_key.encode()).digest()
|
|
return raw_key, prefix, key_hash
|
|
|
|
|
|
def hash_api_key(raw_key: str) -> bytes:
|
|
return hashlib.sha256(raw_key.encode()).digest()
|
|
|
|
|
|
def verify_api_key(raw_key: str, stored_hash: bytes) -> bool:
|
|
return hashlib.sha256(raw_key.encode()).digest() == stored_hash
|