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

Keycloak проверил пароль и выдал пользователю токен. Но дальше встаёт неприятный вопрос: а что этому пользователю вообще можно? Сам по себе токен не запрещает ничего — он лишь говорит, кто пришёл и с какими ролями. Превратить «у тебя есть роль» в «тебе сюда нельзя» — задача вашего сервиса, и Keycloak за вас её не сделает.

В FastAPI у этой задачи нет готового каркаса вроде Spring Security: нет аннотаций, нет конвертера ролей из коробки, нет hasRole. Есть зависимости (Depends), объект запроса и ваш код. Это проще, чем кажется, и честнее: каждая проверка видна в объявлении маршрута и в сценарии.

Обязательно

Аутентификация и авторизация — это разные вещи

  • Аутентификация — «кто ты?». Keycloak узнал пользователя и выдал токен; сервис проверил подпись и понял, что токен настоящий.
  • Авторизация — «что тебе можно?». Создать заказ, посмотреть чужой профиль, зайти в админку.

Keycloak отвечает за первую часть и кладёт в токен роли. Вторую — «с такими ролями сюда можно, а сюда нельзя» — решает сервис. Дальше вся статья про неё. Проверка подписи по JWKS, iss и aud разобрана в статье про проверку токенов в FastAPI; здесь считаем, что токен уже проверен.

Где роли лежат внутри токена

Keycloak различает realm-роли (видны всем приложениям realm: customer, admin) и client-роли (привязаны к одному приложению: order-manager в сервисе заказов). Лежат они в разных местах токена, и это ключевой факт, на котором спотыкаются чаще всего:

Keycloak: роль customer у пользователя выдача токена access_token: realm_access.roles = [customer] разбор claims после проверки подписи зависимость principal: Principal(sub, roles) 403 без роли require_roles("customer") на маршруте

Роль назначают в Keycloak, она едет в токене внутри realm_access.roles, зависимость переводит её в Principal, и маршрут проверяет её явно.

{
  "sub": "a1b2c3d4-...",
  "preferred_username": "ivan",
  "realm_access": {
    "roles": ["customer", "premium"]
  },
  "resource_access": {
    "orders-service": {
      "roles": ["order-manager"]
    }
  },
  "scope": "openid profile email"
}

realm_access.roles — плоский список realm-ролей. resource_access.<client>.roles — client-роли, сгруппированные по идентификатору client, тому самому clientId из настроек Keycloak, а не по отображаемому имени. scope — не про роли пользователя, а про то, что разрешено самому приложению; к нему вернёмся. sub — стабильный идентификатор пользователя, он понадобится для ABAC.

Если сервис читает только realm_access, он никогда не увидит client-роли — и откажет там, где доступ должен быть.

Шаг 1: разбор claims

PyJWT проверяет подпись и стандартные поля, а про realm_access ничего не знает — это формат Keycloak, не часть стандарта. Результат jwt.decode — обычный словарь, и свои поля из него достают явно:

import jwt
from jwt import PyJWKClient

jwks = PyJWKClient(f"{settings.issuer}/protocol/openid-connect/certs", cache_keys=True)


def decode_token(raw: str) -> dict:
    try:
        key = jwks.get_signing_key_from_jwt(raw).key
        return jwt.decode(
            raw,
            key,
            algorithms=["RS256"],
            audience=settings.client_id,
            issuer=settings.issuer,
            options={"require": ["exp", "sub"]},
        )
    except jwt.PyJWTError as e:
        raise Unauthorized("UNAUTHENTICATED", "токен не принят") from e

audience= здесь не украшение: подпись говорит только «токен выпустил наш Keycloak», но не «он выписан нам». Токен соседнего приложения того же realm — с настоящей подписью и ролью admin внутри — пройдёт разбор, если аудиторию не проверять. У PyJWT на этот счёт есть собственная страховка: если в токене есть aud, а в decode аудитория не передана, библиотека откажет с InvalidAudienceError. Чинить это через options={"verify_aud": False} — значит выключить защиту; чинят передачей своего client_id.

Шаг 2: переводчик — Principal в зависимости

Spring переводит роли в GrantedAuthority; в Python такого понятия нет, и его заменяет небольшой неизменяемый объект, который собирает зависимость:

from dataclasses import dataclass, field
from typing import Annotated

from fastapi import Depends, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer


@dataclass(frozen=True)
class Principal:
    subject: str
    roles: frozenset[str] = field(default_factory=frozenset)
    scopes: frozenset[str] = field(default_factory=frozenset)

    def has_role(self, role: str) -> bool:
        return role in self.roles


def principal_from_claims(claims: dict, client_id: str) -> Principal:
    realm_roles = claims.get("realm_access", {}).get("roles", [])
    client_roles = claims.get("resource_access", {}).get(client_id, {}).get("roles", [])
    return Principal(
        subject=claims["sub"],
        roles=frozenset([*realm_roles, *client_roles]),
        scopes=frozenset(claims.get("scope", "").split()),
    )


bearer = HTTPBearer(auto_error=False)


async def current_principal(credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> Principal:
    if credentials is None:
        raise Unauthorized("UNAUTHENTICATED", "нужен токен")
    return principal_from_claims(decode_token(credentials.credentials), settings.client_id)


PrincipalDep = Annotated[Principal, Depends(current_principal)]

Здесь принято два решения, которые стоит проговорить. Realm- и client-роли складываются в одно множество — для проверки доступа не важно, где роль лежала; важно, что она есть. И client-роли читаются только для своего client: роль admin из resource_access чужого приложения в множество не попадает, иначе менеджер соседнего сервиса станет администратором вашего.

Цепочка .get(..., {}).get(...) на отсутствующих ключах возвращает пустой список, и отдельной проверки не нужно. auto_error=False у HTTPBearer нужен, чтобы отсутствие заголовка превращалось в вашу ошибку с вашим кодом, а не в {"detail": "Not authenticated"} от библиотеки. Зависимость с одним и тем же вызываемым объектом FastAPI вычисляет один раз на запрос и переиспользует: токен разбирается однажды, сколько бы зависимостей на него ни ссылалось.

Префикса ROLE_ в Python нет, и целый класс ошибок Spring здесь не существует. Зато есть свой: сравнение строк точное, Admin и admin — разные роли, и имя роли в коде должно совпадать с именем в Keycloak буква в букву.

Шаг 3: где физически стоит проверка

На маршруте проверяют роль. Это ответ на вопрос «доступна ли такая ручка этому типу пользователя»; он не зависит от данных, стоит дёшево и срабатывает до похода в базу. В FastAPI это зависимость на роутере — и это же карта доступа, которую видно одним взглядом:

from fastapi import APIRouter


def require_role(role: str):
    async def check(principal: PrincipalDep) -> Principal:
        if not principal.has_role(role):
            raise Forbidden("ACCESS_DENIED", "недостаточно прав")
        return principal

    return check


customer_api = APIRouter(prefix="/api/v1", dependencies=[Depends(require_role("customer"))])
admin_api = APIRouter(prefix="/api/v1/admin", dependencies=[Depends(require_role("admin"))])


@customer_api.post("/orders", status_code=201)
async def create_order(body: CreateOrder, principal: PrincipalDep, create: CreateOrderDep) -> OrderResponse: ...


@customer_api.get("/orders/{order_id}")
async def get_order(order_id: UUID, principal: PrincipalDep, orders: OrdersDep) -> OrderResponse: ...


@admin_api.delete("/orders/{order_id}", status_code=204)
async def delete_order(order_id: UUID, principal: PrincipalDep, delete: DeleteOrderDep) -> None: ...


app.include_router(customer_api)
app.include_router(admin_api)

Отказ — исключения Unauthorized и Forbidden вашей иерархии, которые единый обработчик превращает в 401 и 403 в том же формате, что и 404, — об этом статья про единый обработчик ошибок. Обработчику внутри группы principal нужен не для проверки роли (она уже прошла), а чтобы передать subject дальше в сценарий.

В сценарии проверяют то, для чего нужны данные: владение записью, её состояние, лимиты. Эту проверку нельзя поднять на маршрут: там объекта ещё нет, а читать его дважды значит проверить одно состояние, а изменить другое.

Чего делать не надо — ставить одну и ту же проверку в оба места. Через полгода правило поменяют в одном, и какое сработает, будет зависеть от того, кто кого вызвал. Роль — на входе, данные — внутри, каждая проверка в одном месте. И отдельно: обработчик сообщения из очереди и задача по расписанию через маршруты не ходят, поэтому правила, обязанные действовать всегда, живут там, где выполняется операция.

RBAC: доступ по ролям

require_role выше и есть RBAC: «у тебя есть роль admin → пускаем». Он отвечает на вопрос «какому типу пользователей вообще можно сюда?» — контроль на уровне действия, не записи. RBAC легко скажет «редактировать заказы может любой customer», но в принципе не способен проверить, свой ли это заказ.

ABAC: когда одной роли мало

У Ивана роль customer, и RBAC разрешает любому customer редактировать заказы. Иван открывает заказ Петра и меняет адрес доставки на свой. Роль правильная — RBAC пропустит. Доступ при этом совершенно неправильный.

ABAC принимает решение по атрибутам: кто владелец записи, кто пришёл, в каком состоянии запись. Самый частый случай — проверка владения по sub:

class ChangeAddress:
    def __init__(self, orders: OrderRepository) -> None:
        self.orders = orders

    async def handle(self, order_id: UUID, address: Address, actor: str) -> None:
        order = await self.orders.by_id(order_id)
        if order.buyer_id != actor:
            raise Forbidden("NOT_YOUR_ORDER", "это не ваш заказ")
        await self.orders.save(order.with_address(address))

В сценарий приходит строка actor, а не Principal и не токен: достать sub из зависимости — работа обработчика, там веб-слой и заканчивается. Иначе сценарий привязан к HTTP и токену Keycloak, и его не вызвать ни из фоновой задачи, ни из теста без сборки фальшивого токена.

Для чтения лучше спросить базу сразу с владельцем: by_id_and_buyer(order_id, actor) и NoResultFound → 404. Тогда существование чужой записи не выдаётся, и проверка не размазывается по коду. Для списков — только фильтр в запросе (.where(Order.buyer_id == actor)): загрузить всё и отфильтровать в Python — значит прочитать чужое, сломать постраничную выдачу и платить за это на каждом запросе.

Обход для администратора — в одном месте

Правило «у роли admin проверка владения не применяется» в коде расползается на десять if principal.has_role("admin"), и в одном из них через полгода ошибутся. Держат его там же, где живёт сама проверка:

class OrderAccess:
    def __init__(self, orders: OrderReader, audit: AuditLog) -> None:
        self.orders = orders
        self.audit = audit

    async def ensure_can_read(self, order_id: UUID, principal: Principal) -> None:
        if principal.has_role("admin"):
            await self.audit.record(principal.subject, "order.read", str(order_id))
            return
        if not await self.orders.owned_by(order_id, principal.subject):
            raise NotFound("ORDER_NOT_FOUND", "заказ не найден")

Побочная польза: тест «администратор видит чужой заказ, покупатель нет» пишется на этот класс с fake-репозиторием, без роутера и базы. Две оговорки: обход нужен не всякому административному действию (прочитать чужой заказ по жалобе — да, менять чужую карточку от чужого имени — нет), и обход всегда оставляет запись в журнале — разбор в статье про журнал действий.

RBAC и ABAC работают вместе

Их не противопоставляют, а складывают: роль отсеивает грубо и рано, владение уточняет на уровне записи. Порядок ради экономии: проверить строку в множестве стоит ничего и срабатывает до базы; дорогая проверка владельца выполняется только для тех, кто прошёл первый фильтр.

При чём тут scope

scope — не про пользователя, а про приложение, которое действует от его имени: «это приложение может читать мои заказы, но не управлять ими». Keycloak отдаёт его строкой через пробел, principal_from_claims разложил её в множество, а у FastAPI для него есть отдельный механизм — Security с перечнем областей:

from fastapi import Security
from fastapi.security import SecurityScopes


async def with_scopes(security_scopes: SecurityScopes, principal: PrincipalDep) -> Principal:
    missing = set(security_scopes.scopes) - principal.scopes
    if missing:
        raise Forbidden("INSUFFICIENT_SCOPE", f"нет прав: {', '.join(sorted(missing))}")
    return principal


@public_api.get("/orders")
async def list_orders(principal: Annotated[Principal, Security(with_scopes, scopes=["orders:read"])]) -> list[OrderResponse]: ...

SecurityScopes получает области из объявления маршрута, а не из токена, поэтому одна зависимость обслуживает все маршруты с разными требованиями. Для обычного доступа внутри своих сервисов почти всегда хватает ролей. Scope нужен в сценариях с внешними клиентами и публичным API, где важно ограничить именно то, на что согласился пользователь.

Сколько ролей заводить

Роль — ярлык, который создаётся в Keycloak парой кликов, и возникает соблазн плодить их: customer-premium, customer-trial, seller-pro, junior-admin. Через полгода два десятка ролей, и проверки превращаются в has_any_role на пол-экрана.

Здоровая дисциплина обратная: ролей мало и стабильно — customer, seller, admin, system для вызовов между сервисами. Новая роль — сигнал остановиться: точно ли это новый тип пользователя? «Premium» — атрибут обычного customer (есть активная подписка), и проверяется он по данным, как владение. «Junior-admin только смотрит» — набор прав внутри admin, а не новая роль. Правило: роль отвечает «кто это за пользователь», а всё, что звучит как «а ещё у него есть/нет свойства», — атрибут, территория ABAC.

Имя роли — соглашение всей системы, а не одного сервиса. Если в одном сервисе покупатель customer, а в соседнем buyer, проверка однажды не найдёт роль и откажет молча. Имена заводят один раз, в одном realm, и держат в одном месте — например, в StrEnum общего пакета.

Каждый маршрут обязан иметь проверку

Самое важное правило темы и то, про которое легче всего забыть. Маршрут вне защищённого роутера открыт всем — и адрес вроде /admin/orders/{id}/refund сам по себе ничего не защищает: это строка. Если возврат денег объявлен на app напрямую, а не на admin_api, его инициирует любой, у кого есть токен, а если и current_principal мимо — вообще любой.

В Python нет аннотаций, по которым это проверил бы ArchUnit, но есть кое-что честнее: обойти все маршруты и дёрнуть каждый без токена.

import pytest
from fastapi.testclient import TestClient

from app.main import app

PUBLIC = {("GET", "/health/live"), ("GET", "/health/ready"), ("GET", "/metrics")}


def all_routes() -> list[tuple[str, str]]:
    return [(method.upper(), path) for path, operations in app.openapi()["paths"].items() for method in operations]


@pytest.mark.parametrize(("method", "path"), all_routes())
def test_route_requires_token(method: str, path: str):
    if (method, path) in PUBLIC:
        pytest.skip("публичный маршрут")
    client = TestClient(app)

    response = client.request(method, path.replace("{order_id}", "0192f1e4-0000-7000-8000-000000000001"))

    assert response.status_code == 401, f"{method} {path} без токена ответил {response.status_code}"

Схема OpenAPI перечисляет все зарегистрированные маршруты с методами, и новый маршрут попадает в тест сам, без правки; единственная оговорка — маршруты с include_in_schema=False в неё не входят, и для них список публичных дополняют руками. Список публичных путей явный и короткий — добавить в него что-то означает осознанно открыть маршрут, и это видно на ревью. Тест ловит и забытый роутер, и маршрут, по ошибке объявленный на app напрямую.

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

  • Ищут роли не в том claim. Realm-роли — в realm_access.roles, client-роли — в resource_access.<clientId>.roles, и ключ — идентификатор client, а не его отображаемое имя.
  • Берут client-роли всех приложений. Цикл по claims["resource_access"].values() складывает в множество роли чужих сервисов; читают только свой client_id.
  • Доверяют ролям без проверки подписи. Роли в JWT — текст внутри JSON. jwt.decode(token, options={"verify_signature": False}) годится только для отладки; в сервисе — decode с ключом из JWKS.
  • Глушат проверку аудитории. options={"verify_aud": False} убирает InvalidAudienceError, но вместе с ней и защиту от токена соседнего приложения; передают audience=client_id.
  • Сравнивают роли без учёта регистра или с пробелами. Имя в коде и в Keycloak совпадают буква в букву, и это проверяет тест на principal_from_claims.
  • Полагаются на RBAC там, где нужен ABAC. Роль customer не гарантирует, что заказ — его.
  • Берут preferred_username вместо sub. Логин можно сменить, sub — нет; владение сравнивают по sub.

Разбор на маркетплейсе: где RBAC, а где ABAC

Возьмём маркетплейс с ролями buyer, seller, moderator, dispute-operator, finance — пять, и это весь каталог.

ДействиеРоль решает?Что ещё нужно проверить
Оформить заказда, buyerничего
Посмотреть свой заказнетвладелец заказа = sub
Отменить заказнетвладелец и статус: после отправки нельзя
Создать карточку товарада, sellerничего
Изменить карточкунетпродавец карточки = sub
Одобрить карточкуда, moderatorничего
Решить спорда, dispute-operatorспор взят этим оператором
Подтвердить выплатуда, financeсумма выше лимита — второй подтверждающий

Действия сотрудников почти целиком закрываются ролью, действия покупателей и продавцов — почти никогда: сотрудник работает со всем потоком, покупатель — только со своим.

Отмена заказа показывает, зачем различать два отказа:

class CancelOrder:
    async def handle(self, order_id: UUID, actor: str) -> None:
        order = await self.orders.by_id(order_id)
        if order.buyer_id != actor:
            raise Forbidden("NOT_YOUR_ORDER", "чужой заказ")
        if not order.cancellable():
            raise Conflict("ORDER_IN_DELIVERY", "заказ уже в доставке, это возврат, а не отмена")
        await self.orders.save(order.cancelled())

Первый отказ — про доступ (403), второй — про состояние (409). Покупателю в интерфейсе нужно показать разные вещи, и смешивать их в один Forbidden не стоит. Выплата продавцу устроена так же: роль finance — вход в проверку, а правило «второй подтверждающий не тот же человек» — атрибуты. И общее: у сотрудников права широкие, поэтому каждое их действие пишется в журнал.

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

Глубже: Authorization Services: когда логика доступа уезжает в Keycloakрасширенное

У Keycloak есть третий способ: Authorization Services. На client включают «Authorization», описывают ресурсы, области действий, политики и разрешения, а сервис спрашивает у Keycloak «можно ли этому пользователю cancel для order:42» и получает токен с разрешениями. Сверху лежит UMA 2.0: пользователь сам делится ресурсом с другим.

Почему для большинства сервисов на Python это не берут: логика доступа уезжает из кода в настройки другого сервиса, её не видно в ревью и не покрывают тесты сценария; проверка «владелец ли заказа» требует, чтобы Keycloak знал о каждом заказе, то есть о миллионах строк вашей базы; появляется сетевой вызов на каждое решение. Библиотека python-keycloak умеет ходить в этот API (KeycloakOpenID.uma_permissions), но policy enforcer, который сам бы вклинивался в маршруты, в ней не предусмотрен — и это тоже сигнал.

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

Коротко

  • Аутентификация — «кто ты» (Keycloak), авторизация — «что тебе можно» (сервис). Путь роли: назначили → попала в токен → сервис проверил подпись и aud → разобрал claims → собрал Principal → сверил с правилом.
  • Realm-роли лежат в realm_access.roles, client-роли — в resource_access.<clientId>.roles только своего client; обе складываются в одно множество неизменяемого Principal, который собирает зависимость один раз на запрос.
  • Префикса ROLE_ в Python нет, зато сравнение строк точное: имя роли в коде совпадает с Keycloak буква в букву; PyJWT сам требует audience=, и глушить это через verify_aud нельзя.
  • RBAC — require_role в dependencies роутера, отказ через свои исключения и единый обработчик; ABAC — владение по sub в сценарии, куда приходит строка actor, а не токен.
  • Для чтения — запрос сразу с владельцем и 404 для чужого; для списков — фильтр в SQL, а не отбор после выборки; обход для администратора — в одном классе OrderAccess с записью в журнал.
  • Scope — права приложения, не пользователя: Security(..., scopes=[...]) и SecurityScopes для публичного API; каталог ролей короткий и стабильный, «premium» и «только смотреть» — атрибуты, не роли.
  • Каждый маршрут обязан быть на защищённом роутере: тест по app.openapi()["paths"] дёргает все маршруты без токена и ждёт 401 для всего, что не в явном списке публичных.
  • Отказ по доступу (403) и отказ по состоянию (409) — разные ответы; Authorization Services и UMA берут только когда права правит администратор заказчика.

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