Когда в API появляются эндпоинты вида GET /orders/{id}, возникает вопрос: а может ли любой авторизованный пользователь читать чужие заказы? Ролевой проверки (RBAC) для этого недостаточно — нужен второй слой.
Проблема: RBAC не защищает от горизонтального перебора
Представьте: у вас есть роль customer, которая разрешает вызывать GET /orders/{id}. Пользователь customer-42 знает свой заказ order-100 и просто меняет число в URL на order-101 — это заказ customer-99. RBAC пропустит запрос, потому что роль есть. Ресурс окажется чужим — но никто этого не проверил.
Роль у обоих запросов одна и та же, поэтому RBAC их не различает; чужой заказ отсекает только сверка владельца после загрузки.
Такая атака называется IDOR (Insecure Direct Object Reference) — прямое обращение к чужому объекту по идентификатору. Это одна из самых частых уязвимостей в API.
ABAC (Attribute-Based Access Control) — контроль доступа на основе атрибутов — решает именно эту проблему. RBAC говорит «роль customer может вызывать этот endpoint», ABAC добавляет «но только к своим ресурсам». Вместе они дают полноценную защиту.
Что такое Principal
В JWT-токене пользователя всегда есть поле sub — уникальный идентификатор пользователя. Когда FastAPI извлекает токен из запроса, удобно завернуть эти данные в датакласс:
# adapters/in/http/security.py
from dataclasses import dataclass
@dataclass(frozen=True)
class Principal:
sub: str # UUID или числовой id пользователя из JWT
roles: list[str]
def is_admin(self) -> bool:
return "admin" in self.roles
principal.sub — это строка. При сравнении с order.customer_id типы должны совпадать: если в базе хранится UUID, приводите явно, иначе сравнение не сработает.
Способ 1: AccessPolicy для простых проверок
Если нужна просто проверка «этот ресурс принадлежит текущему пользователю», хорошо подходит отдельный класс AccessPolicy. Это обычный сервис, который инжектируется через Depends в роутер.
# application/access_policy.py
from fastapi import Depends, HTTPException
from adapters.in.http.security import Principal, principal as get_principal
from domain.order.order_repository import OrderRepository
class OrderAccessPolicy:
def __init__(
self,
repo: OrderRepository = Depends(),
p: Principal = Depends(get_principal),
):
self._repo = repo
self._principal = p
def _load(self, order_id: str):
order = self._repo.find_by_id(order_id)
if order is None:
raise HTTPException(status_code=404, detail="not found")
return order
def can_view(self, order_id: str):
order = self._load(order_id)
if not self._principal.is_admin() and order.customer_id != self._principal.sub:
raise HTTPException(status_code=403, detail="forbidden")
return order
def can_edit(self, order_id: str):
return self.can_view(order_id)
Роутер вызывает политику и получает уже проверенный объект:
# adapters/in/http/order_router.py
from fastapi import APIRouter, Depends
from application.access_policy import OrderAccessPolicy
from adapters.in.http.security import require_roles
router = APIRouter(prefix="/orders")
@router.get("/{order_id}")
async def get_order(
order_id: str,
policy: OrderAccessPolicy = Depends(),
_: None = Depends(require_roles("customer", "admin")),
):
order = policy.can_view(order_id)
return OrderResponse.from_domain(order)
AccessPolicy — единая точка ownership-логики для всех эндпоинтов агрегата. Если завтра бизнес-правило изменится (например, добавится совладелец), меняете одно место, а не каждый роутер.
Способ 2: проверка внутри Handler
Для команд, изменяющих состояние (например, отмена заказа), ABAC удобнее разместить прямо в Handler. Handler владеет транзакцией, загружает агрегат с блокировкой и проверяет бизнес-правила — ABAC здесь органично вписывается в тот же поток:
# application/cancel_order_handler.py
from dataclasses import dataclass
from adapters.in.http.security import Principal
from domain.order.order_repository import OrderRepository
from domain.order.errors import OrderNotFound, OrderCannotBeCancelled
from application.errors import ForbiddenError
from application.audit_log import AuditLog, AdminAction
from datetime import datetime, timezone
@dataclass(frozen=True)
class CancelOrderCommand:
order_id: str
principal: Principal
class CancelOrderHandler:
def __init__(self, repo: OrderRepository, audit_log: AuditLog):
self._repo = repo
self._audit_log = audit_log
def handle(self, cmd: CancelOrderCommand) -> Order:
order = self._repo.find_by_id_for_update(cmd.order_id)
if order is None:
raise OrderNotFound(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")
if not order.can_cancel():
raise OrderCannotBeCancelled(order.id, order.status)
order.cancel()
saved = self._repo.save(order)
if cmd.principal.is_admin():
self._audit_log.record(AdminAction(
actor_id=cmd.principal.sub,
action="cancel-order",
resource_type="Order",
resource_id=order.id,
occurred_at=datetime.now(timezone.utc),
metadata={"previous_status": order.previous_status},
))
return saved
Порядок: загрузка с блокировкой → ABAC → бизнес-правило → сохранение → audit. Handler остаётся тонким роутером — роутер передаёт ему команду и ничего не проверяет сам.
Коллекции: фильтр вместо проверки
Для списка проверка «свой или чужой» на каждом элементе уже поздно: чужие строки прочитаны из базы, и остаётся только надеяться, что фильтр ничего не пропустит. Владелец должен быть условием запроса:
# плохо: прочитали всё, отфильтровали в памяти
all_orders = await orders.find_all()
return [o for o in all_orders if o.customer_id == principal.sub]
# хорошо: чужие строки не читаются вовсе
return await orders.find_by_customer(principal.sub, page)
Разница не только в объёме данных: во втором варианте ошибка в фильтре невозможна, потому что фильтра нет, а есть предикат запроса, который покрыт тестом репозитория.
Состояние объекта — тоже атрибут
Владение это не единственный атрибут, по которому принимают решение. Отменить можно только свой заказ и только пока он не отгружен, и второе условие такой же атрибут объекта, как первое:
async def can_cancel(self, order_id: OrderId, principal: Principal) -> Order:
order = await self._orders.find_by_id_and_customer(order_id, principal.sub)
if order is None:
raise NotFound(order_id) # чужой и несуществующий неотличимы
if order.status not in {Status.NEW, Status.PAID}:
raise NotCancellable(order_id, order.status) # состояние проверено на тех же данных
return order
Обе проверки сделаны на одной загруженной строке, и между ними нет окна: загрузка с условием владельца и проверка статуса идут в одной транзакции, а при SELECT FOR UPDATE ещё и под блокировкой.
Когда какой способ выбрать
| Ситуация | Способ |
|---|---|
Простое сравнение: order.customer_id == principal.sub | AccessPolicy |
| Read-эндпоинты без изменения состояния | AccessPolicy |
Write-команды с SELECT FOR UPDATE | Handler-check |
| Ownership + проверка статуса + бизнес-правило вместе | Handler-check |
| Один агрегат, несколько возможных владельцев | Handler-check |
Важно: выбирайте один из двух способов для каждого агрегата, не оба одновременно. Дубликат расходится при изменении бизнес-правил.
Частая ошибка: ABAC в каждом роутере
Антипаттерн — писать проверку владения прямо в каждой функции роутера:
# ПЛОХО — проверка дублируется в каждом эндпоинте
@router.get("/{order_id}")
async def get_order(order_id: str, p: Principal = Depends(get_principal)):
order = repo.find_by_id(order_id)
if order.customer_id != p.sub:
raise HTTPException(status_code=403)
return order
@router.post("/{order_id}/cancel")
async def cancel_order(order_id: str, p: Principal = Depends(get_principal)):
order = repo.find_by_id(order_id)
if order.customer_id != p.sub: # снова то же самое
raise HTTPException(status_code=403)
...
При изменении правил (добавить совладельца, проверять статус ресурса до проверки владения) придётся обходить все эндпоинты. Пропустите один — уязвимость вернётся.
Admin обходит ABAC, но оставляет след
Администратор часто должен видеть любые ресурсы — для поддержки, проверки, расследования. Это нормально, но каждое такое действие должно быть зафиксировано в журнале:
if query.principal.is_admin():
self._audit_log.record(AdminAction(
actor_id=query.principal.sub,
action="view-product",
resource_type="Product",
resource_id=product.id,
occurred_at=datetime.now(timezone.utc),
metadata={"seller_id": product.seller_id},
))
В журнале появится запись: actor_id=admin-7, action=view-product, resource_id=prod-555. Если кто-то злоупотребляет admin-доступом, это видно при аудите.
Проверка is_admin() должна быть во всех ветках — не только для audit, но и чтобы пропустить ownership-проверку. Обычный паттерн: if not principal.is_admin() and resource.owner != principal.sub: raise ForbiddenError(...).
Тест, который ловит IDOR
Самый дешёвый тест безопасности в проекте: два покупателя, заказ одного, запрос от имени другого, ожидание 404 без единого байта чужих данных:
async def test_foreign_order_is_not_found(client: AsyncClient, token_for):
foreign_order = await create_order(customer_id="customer-2")
response = await client.get(f"/orders/{foreign_order.id}", headers={"Authorization": f"Bearer {token_for('customer-1')}"})
assert response.status_code == 404
assert "customer-2" not in response.text
Такой тест пишут на каждый эндпоинт с идентификатором в пути и держат в одном модуле: когда появляется новый ресурс, в модуле добавляют одну строку, а не вспоминают о проверке на инциденте.
Глубже: перебор идентификаторов: 404, UUID и ограничение частотырасширенное
404 вместо 403 для чужого ресурса закрывает утечку факта существования. Он не закрывает перебор: клиент с токеном обычного покупателя может за ночь перебрать /orders/1 … /orders/1000000 и по ответам собрать всё, что проверка пропустит хотя бы раз, а по времени ответа отличить «нет такого» от «есть, но чужой», если первый ответ приходит без похода в базу, а второй после.
Три меры, по возрастанию стоимости. Одинаковый путь для обоих случаев: AccessPolicy загружает ресурс и только потом решает, и 404 для чужого и для несуществующего идут через один код с одним телом. Непоследовательные идентификаторы: UUID в пути делает перебор бессмысленным, а FastAPI отбрасывает невалидные значения до обработчика; с Python 3.14 для новых таблиц берут uuid.uuid7(), он не раскрывает счётчик и дружит с индексом. Лимит частоты по пользователю на чтение по идентификатору: сто 404 подряд от одного sub за минуту это не ошибка клиента, а сканирование, и на него ставят 429 и тревогу; счётчик живёт в Redis, как в статье про ограничение частоты.
Глубже: мультиарендность: tenant_id это не рольрасширенное
Самый частый случай ABAC на практике не «владелец заказа», а «арендатор»: одна установка обслуживает много компаний, и данные одной не должны быть видны другой ни при каких обстоятельствах. Новичок решает это ролью «менеджер компании X», и это не работает: ролей становится столько же, сколько компаний, а забытый фильтр в одном запросе открывает всё.
Арендатор это атрибут: claim tenant_id в токене, поле Principal.tenant_id, и колонка tenant_id в каждой таблице с данными арендатора. Дальше задача одна: фильтр по tenant_id обязан стоять в каждом запросе, и держать его на внимательности нельзя. Два способа сделать это автоматически.
На стороне SQLAlchemy: событие do_orm_execute на сессии добавляет with_loader_criteria(TenantScoped, lambda cls: cls.tenant_id == current_tenant.get()) ко всем ORM-запросам по моделям с общим миксином TenantScoped; значение берут из ContextVar, который выставляет зависимость principal. Сырые text()-запросы под это правило не попадают, и это его слабое место.
На стороне PostgreSQL: row-level security. ALTER TABLE orders ENABLE ROW LEVEL SECURITY и политика USING (tenant_id = current_setting('app.tenant_id')::uuid), а приложение в начале транзакции выполняет SET LOCAL app.tenant_id = :tenant. Тогда фильтр работает и для ORM, и для text(), и для отчёта, который кто-то напишет через год. Цена: SET LOCAL в каждой транзакции и роль базы без права обходить политики.
Тест на границу обязателен и прост: создать данные как арендатор A, запросить как B, ожидать пустой список, для каждого эндпоинта списка и каждого эндпоинта по идентификатору.
Коротко
- RBAC говорит «роль разрешает вызов», ABAC добавляет «только к своим ресурсам» — оба слоя нужны вместе.
- Без ABAC RBAC-эндпоинт открыт для IDOR: любой пользователь с нужной ролью перебирает чужие объекты по id.
- Два способа разместить ABAC: AccessPolicy (для read и простых ownership) и Handler-check (для write с транзакцией и бизнес-правилами).
- Ownership-логика — в одном месте, не дублируется по роутерам.
principal.sub— строка из JWT; при сравнении с id из базы следите за типами.- Admin обходит ABAC, но каждое его действие записывается в audit log.
- Перебор идентификаторов режут одинаковым
404,UUIDв пути и лимитом на404по пользователю; арендатор — атрибутtenant_id, который фильтруется автоматически черезwith_loader_criteriaили row-level security PostgreSQL, не роль.
Что пощупать
Владение ресурсом во взрослом каталоге практикума remodov/marketplace-system-python: роль из токена отвечает, что продавцу можно вообще, а require_ownership в сценарии - можно ли ему вот этот товар; чужой товар выглядит несуществующим, 404 OWN_PRODUCT_REQUIRED, а не 403, чтобы перебором нельзя было узнать чужие идентификаторы. Та же проверка стоит перед выдачей подписанной ссылки на загрузку фото.
Код: ownership.py, request_image_upload.py, test_image_upload.py.
Сделаем сами
Ветки step-07-adult-catalog и step-12-tokens-and-files - в обеих чужой товар должен отдавать 404.
Что почитать дальше
- RBAC: маппинг ролей — слой до ABAC.
- Audit admin-команд — обязательная пара к admin-override.
- JWT validation — как получить
Principalиз токена.