5fcb574aca
Test / test (push) Successful in 15s
- Reformatted function signatures in `organization_service.py` and `task_service.py` for better alignment. - Updated import statements across multiple files for consistency and organization. - Enhanced test files by improving formatting and ensuring consistent use of async session factories. - Added type hints and improved type safety in various service and test files. - Adjusted `pyproject.toml` to include configuration for isort, mypy, and ruff for better code quality checks. - Cleaned up unused imports and organized existing ones in several test files.
59 lines
1.7 KiB
Python
59 lines
1.7 KiB
Python
"""Security helpers for hashing passwords and issuing JWT tokens."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Any
|
|
|
|
import jwt
|
|
from app.core.config import settings
|
|
from passlib.context import CryptContext # type: ignore
|
|
|
|
|
|
class PasswordHasher:
|
|
"""Wraps passlib context to hash and verify secrets."""
|
|
|
|
def __init__(self) -> None:
|
|
self._context = CryptContext(schemes=["pbkdf2_sha256"], deprecated="auto")
|
|
|
|
def hash(self, password: str) -> str:
|
|
return self._context.hash(password)
|
|
|
|
def verify(self, password: str, hashed_password: str) -> bool:
|
|
return self._context.verify(password, hashed_password)
|
|
|
|
|
|
class JWTService:
|
|
"""Handles encoding and decoding of JWT access tokens."""
|
|
|
|
def __init__(self, secret_key: str, algorithm: str) -> None:
|
|
self._secret_key = secret_key
|
|
self._algorithm = algorithm
|
|
|
|
def create_access_token(
|
|
self,
|
|
subject: str,
|
|
expires_delta: timedelta,
|
|
claims: Mapping[str, Any] | None = None,
|
|
) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
payload: dict[str, Any] = {
|
|
"sub": subject,
|
|
"iat": now,
|
|
"exp": now + expires_delta,
|
|
}
|
|
if claims:
|
|
payload.update(claims)
|
|
return jwt.encode(payload, self._secret_key, algorithm=self._algorithm)
|
|
|
|
def decode(self, token: str) -> dict[str, Any]:
|
|
return jwt.decode(token, self._secret_key, algorithms=[self._algorithm])
|
|
|
|
|
|
password_hasher = PasswordHasher()
|
|
jwt_service = JWTService(
|
|
secret_key=settings.jwt_secret_key.get_secret_value(),
|
|
algorithm=settings.jwt_algorithm,
|
|
)
|