RBAC в FastAPI: роли, маппинг и защита эндпоинтов

← назад к разделу

В приложении обычно есть разные типы пользователей: покупатели, продавцы, администраторы. Каждый должен попадать только туда, куда ему положено. Ручные if-проверки внутри каждого обработчика — это хаос: логика расползается по всему коду, и легко пропустить незащищённый эндпоинт.

RBAC (Role-Based Access Control, контроль доступа на основе ролей) решает эту задачу системно: роль пользователя определяет, к каким эндпоинтам он вообще может обращаться.

Откуда берётся роль пользователя

Когда пользователь входит через Keycloak или другой OAuth2-провайдер, сервер выдаёт ему JWT-токен. Внутри токена, в секции realm_access.roles, хранится список ролей:

{
  "sub": "user-42",
  "realm_access": {
    "roles": ["customer", "loyalty-member"]
  }
}

Стандартный OAuth2 (без Keycloak) кладёт роли в поле scope как строку через пробел: "scope": "customer read:orders".

Задача приложения — достать эти роли из токена при каждом запросе и сделать их доступными в обработчике.

Principal: объект, который представляет пользователя

Вместо того чтобы работать напрямую с сырым словарём из JWT, удобно завернуть данные о пользователе в отдельный датакласс Principal. Он хранит идентификатор пользователя (sub) и список его ролей:

# adapters/in/http/security.py
from dataclasses import dataclass
from typing import Any

from fastapi import Depends, HTTPException
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
import jwt
from jwt import PyJWKClient

from config import settings

_bearer = HTTPBearer()
_jwks = PyJWKClient(settings.jwks_uri, cache_keys=True, lifespan=300)


@dataclass(frozen=True)
class Principal:
    sub: str
    roles: list[str]

    def is_admin(self) -> bool:
        return "admin" in self.roles


def _roles(claims: dict[str, Any]) -> list[str]:
    realm = claims.get("realm_access", {})
    if isinstance(realm, dict) and "roles" in realm:
        return realm["roles"]
    scope = claims.get("scope", "")
    return scope.split() if scope else []


async def principal(
    creds: HTTPAuthorizationCredentials = Depends(_bearer),
) -> Principal:
    token = creds.credentials
    try:
        key = _jwks.get_signing_key_from_jwt(token).key
        claims = jwt.decode(
            token, key, algorithms=["RS256"],
            audience=settings.audience, issuer=settings.issuer,
        )
    except jwt.PyJWTError as e:
        raise HTTPException(status_code=401, detail="invalid token") from e
    return Principal(sub=claims["sub"], roles=_roles(claims))

Функция _roles умеет читать роли в обоих форматах: Keycloak и стандартный OAuth2. Она всегда возвращает list[str], поэтому остальной код не знает, откуда пришли роли.

Зависимость principal проверяет подпись токена через JWKS (публичные ключи провайдера), проверяет аудиторию и издателя, и только после этого собирает Principal. Невалидный или просроченный токен → 401 Unauthorized.

require_roles: защита эндпоинта

Зная, как получить Principal, несложно написать зависимость, которая проверяет роли:

def require_roles(*roles: str):
    async def dep(p: Principal = Depends(principal)) -> Principal:
        if not set(roles) & set(p.roles):
            raise HTTPException(status_code=403, detail="forbidden")
        return p
    return dep

Если у пользователя нет ни одной из перечисленных ролей → 403 Forbidden. Семантика — OR: достаточно совпасть хотя бы с одной ролью.

Использование в роутере:

# adapters/in/http/order_router.py
from fastapi import APIRouter, Depends
from adapters.in.http.security import Principal, require_roles
from application.create_order_handler import CreateOrderCommand, CreateOrderHandler
from application.get_order_handler import GetOrderByIdQuery, GetOrderHandler
from application.cancel_order_handler import CancelOrderCommand, CancelOrderHandler

router = APIRouter(prefix="/orders")


@router.post("")
async def create_order(
    request: CreateOrderRequest,
    p: Principal = Depends(require_roles("customer")),
    handler: CreateOrderHandler = Depends(),
):
    return handler.handle(CreateOrderCommand(customer_id=p.sub, **request.model_dump()))


@router.get("/{order_id}")
async def get_order(
    order_id: str,
    p: Principal = Depends(require_roles("customer", "admin")),
    handler: GetOrderHandler = Depends(),
):
    return handler.handle(GetOrderByIdQuery(order_id=order_id, principal=p))


@router.post("/{order_id}/cancel")
async def cancel_order(
    order_id: str,
    p: Principal = Depends(require_roles("customer", "admin")),
    handler: CancelOrderHandler = Depends(),
):
    return handler.handle(CancelOrderCommand(order_id=order_id, principal=p))

Важно: эндпоинт без Depends(require_roles(...)) открыт всем, у кого есть валидный токен — даже если он лежит в /admin/ namespace. Наличие роутера с префиксом /admin ничего не защищает само по себе. Каждый эндпоинт декларирует свои роли явно.

Сколько ролей нужно

Хорошая отправная точка — четыре роли:

Роль	Кто	Что делает
`customer`	Конечный пользователь	Создаёт и читает свои заказы, оплачивает
`seller`	Продавец на маркетплейсе	Управляет своими товарами, видит заказы своих товаров
`admin`	Внутренний пользователь	Полный доступ, действия логируются
`system`	Другой сервис	Межсервисные вызовы (Client Credentials или mTLS)

Чем больше ролей, тем сложнее система. Частая ловушка — вводить новую роль там, где нужен атрибут:

«premium-customer» — это не роль, а атрибут пользователя (customer.is_premium), который проверяется в обработчике.
«junior-admin, который только читает» — это не роль, а система прав (permissions), что сложнее и нужна редко.
«B2B-клиенты видят другой каталог» — скорее всего, это вообще другой сервис.

Новая роль нужна, когда появляется принципиально новый тип участника системы, а не когда появляется новая фича.

RBAC и ABAC — два разных вопроса

RBAC отвечает на вопрос: «может ли пользователь с ролью customer вообще обращаться к GET /orders/{order_id}?»

RBAC не отвечает на вопрос: «является ли order-12345 заказом именно этого пользователя?» Это уже ABAC (Attribute-Based Access Control) — нужно загрузить заказ из базы и сравнить order.customer_id с principal.sub.

# RBAC — на уровне эндпоинта: кто вообще имеет право
@router.post("/{order_id}/cancel")
async def cancel_order(
    order_id: str,
    p: Principal = Depends(require_roles("customer", "admin")),
    handler: CancelOrderHandler = Depends(),
):
    return handler.handle(CancelOrderCommand(order_id=order_id, principal=p))


# ABAC — на уровне бизнес-логики: именно ли его этот заказ
class CancelOrderHandler:
    def handle(self, cmd: CancelOrderCommand) -> Order:
        order = self._repo.find_by_id_for_update(cmd.order_id)
        if not cmd.principal.is_admin() and order.customer_id != cmd.principal.sub:
            raise ForbiddenError("order does not belong to current user")
        order.cancel()
        return self._repo.save(order)

Оба слоя нужны: без RBAC любой аутентифицированный пользователь пробует добраться до чужих ресурсов, без ABAC — customer с валидным токеном получает чужие данные.

Тесты

Проверять авторизацию стоит отдельными тестами — не полагаться на то, что «бизнес-логика всё равно упадёт»:

# tests/test_order_router_auth.py
import pytest
from httpx import AsyncClient


@pytest.mark.anyio
async def test_create_order_requires_customer_role(client: AsyncClient, seller_token: str):
    resp = await client.post(
        "/orders",
        json={"product_id": "prod-1", "quantity": 2},
        headers={"Authorization": f"Bearer {seller_token}"},
    )
    assert resp.status_code == 403


@pytest.mark.anyio
async def test_create_order_without_token_returns_401(client: AsyncClient):
    resp = await client.post("/orders", json={"product_id": "prod-1", "quantity": 2})
    assert resp.status_code == 401

401 — токена нет или он невалиден. 403 — токен валиден, но роль не подходит. Эти коды нельзя путать: 401 означает «войдите», 403 — «у вас нет права».

Частые ошибки

Эндпоинт без проверки роли. Если забыть Depends(require_roles(...)), эндпоинт открыт всем, у кого есть токен. Это не «ограниченный доступ», это публичный эндпоинт за токеном.

Много ролей. Десять ролей — признак того, что RBAC пытается делать работу атрибутов или прав доступа. Держите роли минимальными, выносите детали в бизнес-логику.

Проверка владения через роли. if "customer" in p.roles and order.customer_id == p.sub — это ABAC, написанный в роутере. Ему место в обработчике.

Роли разбросаны строками по коду. Константы или Enum вместо строк в нескольких местах — иначе опечатка будет незаметна.

Коротко

RBAC — первый слой авторизации: определяет, какие роли имеют доступ к эндпоинту.
Роли берутся из JWT: realm_access.roles (Keycloak) или scope (стандартный OAuth2).
Один класс Principal хранит sub и roles; одна функция _roles() умеет читать оба формата.
require_roles(*roles) — зависимость FastAPI с OR-семантикой: достаточно одной совпавшей роли.
Каждый эндпоинт декларирует роли явно; отсутствие require_roles = открытый доступ.
Четыре базовые роли: customer, seller, admin, system. Новая роль нужна редко.
RBAC отвечает «кто может», ABAC — «чей ресурс». Оба слоя нужны.
401 — нет/невалидный токен; 403 — нет нужной роли.

Что почитать дальше

JWT validation в FastAPI — как устроена проверка подписи и зачем нужен JWKS-кеш.
ABAC: владение ресурсом — следующий слой после RBAC.
Service-to-service авторизация — роль system, Client Credentials, mTLS.