lauadminandClaude Opus 5.5
2b7a92645a
- CRM_USE_STUB: local runs no longer reach the live CRM. With only the Checkbox stub, a stub receipt would still move the real order to PACKED. Refused in production, same as CHECKBOX_USE_STUB. - scripts/deploy.sh: backup, fast-forward main, build, health check and code rollback on failure. scripts/backup.sh: pg_dump with verification and 14-day rotation (used by cron and deploy.sh). - Gitea Actions CI: ruff + pytest, oxlint + build. - DEPLOY.md runbook; CLAUDE.md rules for safe local development. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
"""Зависимости FastAPI: сессия БД, текущий пользователь, проверка ролей."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from collections.abc import Callable, Coroutine
|
|
from typing import Annotated, Any
|
|
|
|
from fastapi import Depends, HTTPException, status
|
|
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.security import TokenError, decode_access_token
|
|
from app.db.models.user import User, UserRole
|
|
from app.db.session import get_session
|
|
from app.services.checkbox.client import CheckboxClient, get_checkbox_client
|
|
from app.services.crm.client import CrmClient, get_crm_client
|
|
from app.services.task_queue import TaskQueue, get_task_queue
|
|
|
|
bearer_scheme = HTTPBearer(auto_error=False)
|
|
|
|
SessionDep = Annotated[AsyncSession, Depends(get_session)]
|
|
|
|
_UNAUTHORIZED = HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail="Потрібна автентифікація",
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
)
|
|
|
|
|
|
async def get_current_user(
|
|
session: SessionDep,
|
|
credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)],
|
|
) -> User:
|
|
if credentials is None:
|
|
raise _UNAUTHORIZED
|
|
|
|
try:
|
|
payload = decode_access_token(credentials.credentials)
|
|
except TokenError as exc:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
|
detail=str(exc),
|
|
headers={"WWW-Authenticate": "Bearer"},
|
|
) from exc
|
|
|
|
try:
|
|
user_id = uuid.UUID(payload["sub"])
|
|
except (KeyError, ValueError) as exc:
|
|
raise _UNAUTHORIZED from exc
|
|
|
|
# Пользователь читается из БД на каждый запрос, а не берётся из токена:
|
|
# отключённая учётка должна терять доступ немедленно, не дожидаясь
|
|
# истечения access-токена.
|
|
user = await session.get(User, user_id)
|
|
if user is None or not user.is_active:
|
|
raise _UNAUTHORIZED
|
|
|
|
return user
|
|
|
|
|
|
CurrentUser = Annotated[User, Depends(get_current_user)]
|
|
|
|
|
|
def require_roles(*roles: UserRole) -> Callable[..., Coroutine[Any, Any, User]]:
|
|
"""Ограничивает эндпоинт набором ролей.
|
|
|
|
Пример:
|
|
@router.post("/", dependencies=[Depends(require_roles(UserRole.ADMIN))])
|
|
"""
|
|
allowed = set(roles)
|
|
|
|
async def _guard(user: CurrentUser) -> User:
|
|
if user.role not in allowed:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail="Недостатньо прав для цієї дії",
|
|
)
|
|
return user
|
|
|
|
return _guard
|
|
|
|
|
|
# Готовые зависимости под роли из плана.
|
|
require_admin = require_roles(UserRole.ADMIN)
|
|
require_cashier = require_roles(UserRole.ADMIN, UserRole.CASHIER)
|
|
require_any = require_roles(UserRole.ADMIN, UserRole.CASHIER, UserRole.VIEWER)
|
|
|
|
AdminUser = Annotated[User, Depends(require_admin)]
|
|
CashierUser = Annotated[User, Depends(require_cashier)]
|
|
|
|
|
|
CrmClientDep = Annotated[CrmClient, Depends(get_crm_client)]
|
|
|
|
|
|
CheckboxClientDep = Annotated[CheckboxClient, Depends(get_checkbox_client)]
|
|
TaskQueueDep = Annotated[TaskQueue, Depends(get_task_queue)]
|