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

В приложении обычно есть разные типы пользователей: покупатели, продавцы, администраторы. Каждый должен попадать только туда, куда ему положено. Ручные 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, роутер Depends(require_roles("customer")) роль есть → дальше нет → 403 ABAC, обработчик загрузить заказ customer_id == principal.sub? нет → 403 или 404

Роль проверяют на входе, до загрузки данных; владение проверяют в обработчике, когда заказ уже загружен и известно, чей он.

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 вместо строк в нескольких местах — иначе опечатка будет незаметна.

Дополнительно: при первом чтении можно пропустить

Глубже: роль поменяли, а токен живётрасширенное

Роли приезжают внутри токена, и это значит, что отзыв роли вступает в силу не сразу. Администратора уволили в 10:00, его access-токен действует до 10:15, и все проверки require_roles("admin") его пропускают, потому что подпись и exp в порядке.

Первая мера дешёвая: короткий срок жизни access-токена, 5–15 минут, и refresh с проверкой состояния пользователя на стороне IdP. Пятнадцать минут окна для большинства ролей приемлемы.

Когда неприемлемы (блокировка сотрудника, компрометация аккаунта), окно закрывают на своей стороне. Вариант первый: список отозванных sub в Redis с временем жизни, равным сроку токена; principal после проверки подписи делает одно чтение и отказывает, если sub в списке. Вариант второй: версия прав пользователя, permissions_version, в базе и в токене как claim; расхождение означает «права менялись, токен устарел». Оба стоят одного обращения к хранилищу на запрос, и его прячут за кешем на 10–30 секунд, иначе Redis становится узким местом раньше базы.

Чего не делают: не читают роли из базы на каждый запрос вместо токена. Это возвращает сессии на сервере со всеми их издержками и делает IdP лишним.

Глубже: тесты авторизации без IdPрасширенное

Тесты из раздела выше ходят с настоящими токенами, а значит, нужен IdP или подписанный тестовым ключом JWT. Для большинства проверок это лишнее: роутер должен отвечать 403 на роль seller, и откуда взялся Principal, роутеру безразлично.

Зависимость подменяют на время теста:

@pytest.fixture
def as_role(app):
    def _as(*roles: str):
        app.dependency_overrides[principal] = lambda: Principal(sub="user-1", roles=list(roles))
    yield _as
    app.dependency_overrides.pop(principal, None)

Дальше таблица «эндпоинт, метод, роль, ожидаемый код» через pytest.mark.parametrize: ("POST", "/orders", "seller", 403), ("POST", "/orders", "customer", 201), и отдельная строка без подмены, где ожидается 401. Таблица ценна тем, что новый эндпоинт в неё добавляют вместе с кодом, и ревью видит, какие роли он пропускает.

Настоящий токен нужен в одном месте: в тесте самой зависимости principal. Для него поднимают пару ключей в фикстуре, подписывают JWT приватным и подсовывают PyJWKClient ответ с публичным, или кладут публичный ключ прямо в настройки тестового профиля. Так проверяют exp, aud, iss и alg, а роутеры в это не вовлекают.

Коротко

  • 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 — нет нужной роли.
  • Отзыв роли приходит с новым токеном: короткий срок жизни, а для блокировок — список отозванных sub или версия прав; роутеры тестируют подменой principal через dependency_overrides, настоящий JWT нужен только тесту самой зависимости.

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