RBACX — что изменилось за полгода: от простого RBAC/ABAC до ReBAC с ИИ-генерацией политик

[Vladislav2951]

Как-то случайно наткнулся на библиотеку rbacx, и она кажется даже чуть более дружелюбной, чем аналоги.

Я уже несколько месяцев в поиске хорошего решения, которое позволило бы реализовать RBAC + проверку на owner. Очень частая ситуация (у меня), когда нужно реализовать политику вида "админ может всё, пользователь может просматривать определённые ресурсы, но изменять – только свои". Не хочется ради этого делать полноценный ABAC.

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

Пример

# API
@router.patch(
    "/{user_id}",
    summary="Update user",
)
async def update(
    user_id: UUID7,
    update_data: UpdateUserDTO,
    user_srv: UserService = Depends(user_srv),
    identity_ctx: IdentityContext = Depends(
        IdentityContextFactory(business_element_name, Action.UPDATE)
    ), # Получение пользователя и его привелегий
):
    try:
        user = await user_srv.update(user_id, update_data, identity_ctx)
        return user
    except ForbiddenError:
        raise forbidden()
    except NotFoundError:
        raise not_found("User not found")
    except Exception as e:
        raise internal_server_error()

# Сервис
async def update(
    self, id: UUID7, update_inp: PatchUpdateUserInput, identity_ctx: IdentityContext
) -> User:
    # Здесь возникает нарушение SRP (проверка авторизации)
    await self.access_srv.check_access(
        identity_ctx.current_user,
        identity_ctx.business_element_name,
        identity_ctx.action,
        id, # Owner ID
    ) # raises ForbiddenError

    # Если текущий пользователь != владелец, то далее код не будет выполнен

    async with self._db_transaction_factory() as t:
        user = await t.user_repo.get_one(id)
        if not user:
            raise NotFoundError(f"User {id} not found")

        return await t.user_repo.update(id, update_inp)

В случае с ресурсом User идентификатор владельца очевиден. Но если бы это был, например, ресурс Post, то проверку пришлось бы выполнять только после получения post, и это также нарушение SRP.

Мне как начинающему разработчику в документации хотелось бы видеть чуть более подробные примеры подобной реализации. Пока что это основная преграда в использовании этой и других библиотек авторизации.

[Cheater121]

Привет! Спасибо за положительный фидбек, рад что либа показалась более дружелюбной 😊

Насколько понял, owner_id живёт в БД, а в URL/запросе его нет — правильно?

Если да, то надо уточнить пару моментов, чтобы дать адекватный ответ.

Вас интересует минимизация обращений к БД, или скорее чистота архитектуры? Тут можно решить двумя способами — либо две проверки (первая: «юзер вообще может патчить этот тип ресурса?», потом вторая уже с owner_id), но это два вызова guard и одно обращение к БД. Или интересует кейс, где ресурс один раз достаётся на запрос — кладётся куда-то в зависимость/мидлвэр — и дальше по коду уже не нужно его тащить повторно?

В итоге у вас финальная цель такая: загрузить ресурс, достать из него owner_id, положить в Resource(attrs={"owner_id": ...}) и выполнить проверку условия subject.id == resource.attrs.owner_id — я правильно понимаю?

Чтобы в политике оно выглядело как-то так:

# Юзер может изменять только свой ресурс
        {
            "id": "user_update_own",
            "effect": "permit",
            "actions": ["update"],
            "resource": {"type": "user"},
            "roles": ["user"],
            "condition": {
                "==": [
                    {"attr": "subject.id"},
                    {"attr": "resource.attrs.owner_id"}
                ]
            }
        }

Правильно?

[Vladislav2951]

Добрый день! Спасибо за отклик!

owner_id живёт только в БД вместе с запрашиваемым ресурсом. Ресурс Post может иметь user_id (по сути это и есть owner_id, а для ресурса User его ID – это тоже owner_id).

Задача: реализовать авторизацию так, чтобы пользователь без прав администратора мог редактировать только самого себя (например, биографию, email, что угодно), но не мог редактировать другого пользователя; аналогично с Post – можно редактировать свои посты, но нельзя – чужие.

Пример схем ресурсов

class User(BaseModel):
    id: UUID7 # Это owner_id
    email: EmailStr
    password_hash: SecretStr
    nickname: NicknameStr
    bio: Optional[BioString]
    deleted_at: Optional[datetime] = None
    
class Post(BaseModel):
    id: UUID7
    title: TitleStr
    body: BodyString
    authored_by: NicknameStr
    user_id: UUID7 # Это owner_id
    created_at: datetime

Пример типового запроса на редактирование поста: PATCH /posts/{post_id}. Здесь post_id – это ID ресурса, но НЕ owner_id.

Меня интересует и минимизация обращений к БД, и чистота архитектуры))) В качестве костыля (а может, это вовсе и не костыль) я нашёл решение: для всех объектов, у которых есть владелец, сделать отдельный метод в соответствующем репозитории (и сервисе) – get_owner_id(self, id: UUID7) -> Optional[UUID7]. Я сделал отдельный middleware, который проверяет авторизацию для любого объекта, а также реализовал кастомный middleware для каждого "ownable" ресурса, который дополнительно дергает метод get_owner_id (правда, пока без сторонних библиотек).

Пример реализации данных middleware

# Базовый (универсальный) чекер, без проверки владения
class PermissionChecker:
    def __init__(self, business_element_name: str, action: Action):
        self.business_element_name = business_element_name
        self.action = action

    async def __call__(
        self,
        access_srv: AccessService = Depends(access_srv),
        current_user: User = Depends(get_current_user),
    ):
        try:
            await access_srv.check_access(
                current_user, self.business_element_name, self.action
            )
        except ForbiddenError:
            raise forbidden()

# Выдёргивает post_id из пути (posts/{post_id})            
async def get_post_id(post_id: UUID7 = Path()) -> UUID7:
    return post_id


# Конкретный чекер
class PostPermissionChecker:
    def __init__(self, business_element_name: str, action: Action):
        self.business_element_name = business_element_name
        self.action = action

    async def __call__(
        self,
        access_srv: AccessService = Depends(access_srv),
        post_srv: PostService = Depends(post_srv),
        current_user: User = Depends(get_current_user),
        post_id: UUID7 = Depends(get_post_id),
    ):
        try:
            owner_id = await post_srv.get_owner_id(post_id)
            await access_srv.check_access(
                current_user, self.business_element_name, self.action, owner_id
            )
        except ForbiddenError:
            raise forbidden()
        except Exception:
            raise internal_server_error()

# Пример использования кастомного
@router.patch(
    "/{post_id}", # Полный маршрут: /posts/{post_id}
    summary="Update post",
    response_model=DataResponse[PostPublic],
    responses={status.HTTP_200_OK: {"description": "Success"}},
    dependencies=[
        Depends(PostPermissionChecker(business_element_name, Action.UPDATE))
    ],
)
async def update(
    post_id: UUID7,
    update_data: PostUpdateDTO,
    post_srv: PostService = Depends(post_srv),
):
  pass # Далее просто передаём данные в сервис (авторизация должна быть проведена)

Плюсы:

• Ресурс читается (не полностью, только его owner_id) один раз на уровне API.

• Авторизация не протекает на уровень бизнес-логики, а остаётся в middleware на уровне api.

• Проверка аутентификации проводится отдельно – для всей группы маршрутов (в примере выше не показана).

• Чуть более централизованное определение политик доступа к ресурсу (в middleware).

• Минимально необходимое и достаточное количество чтений ресурса пред выдачей ошибки 403.

Но у этого решения есть несколько минусов, которые мне кажутся таковыми:

• Много дублирования кода (почти однотипные middleware). Возможно, решается через наследование, тут я пока не придумал изящного решения.

• Неочевидный side-эффект middleware: если вместо PostPermissionChecker использовать стандартный PermissionChecker, то никакой проверки owner_id не будет, а если использовать "чужой" middleware (например, UserPermissionChecker), то – будет произведена совершенно другая проверка. Возможно, поможет интерфейс или протокол, но тут тоже пока не знаю, как подступиться.

• Неочевидная зависимость: если в маршруте /posts/{post_id} изменить имя переменной пути post_id, например, на id, то при вызове данной ручки будет ошибка 500 в рантайме (в методе get_post_id).

• Сложность модификации правил. Например, правило "пользователь может видеть только свои посты, администратор может видеть все" приведёт к "протеканию" ответственности к конкретному маршруту, т.к. придётся принудительно добавлять фильтр по owner_id для пользователя, и убирать этот фильтр для админа.

В целом, ощущается, что за эти пару дней я стал чуть ближе к некому идеальному решению, но пока неизвестно, насколько.

Конечная цель – сделать проверку авторизации самодостаточной изолированной операцией и не допустить "протечек" ответственности.

[Cheater121]

Ну путь правильный, с универсальной зависимостью - get_owner_id. Хотя я бы сделал чуть шире - get_resource общий как вспомогательную функцию (прочитаете ниже - поймете почему даже это не обязательно).

Будто бы сейчас смешаны два уровня:

1. получение owner_id

2. принятие решения о доступе

Именно из-за этого появляются:

• дублирование мидлвэра,

• хрупкость (post_id vs id),

• риск забыть “правильный” checker,

• протекание правил в маршруты.

• И тп)

Главная мысль, как это исправить - тут вообще НЕ нужен отдельный мидлвэр на каждый ресурс.

Я бы сделал так:

• один универсальный dependency/checker,

• единый build_env,

• передача owner_id в resource.attrs,

• ну и собственно настроенная политика в rbacx.

У меня в либе уже есть под это нужные болванки - require_access под фастапи и пример энв билдера. По сути надо его адаптировать, вписав туда чисто свою логику получения Subject, Resource, Action, Context (оно именуется в доках SARC для краткости).

Middleware (вообще зависимость, а не мидлвэр, но про это ниже) должен:

• собрать SARC, сам или заюзав энв-билдер;

• вызвать guard.

Архитектурная проблема начинается тут:

PostPermissionChecker UserPermissionChecker CommentPermissionChecker

И ТпПермишнЧекер...это можно было бы решить фабрикой конечно, но я немного скорректировал бы принципиально подход.

По сути нужно перейти от:

“middleware знает как проверять ownership”

к:

“middleware только собирает env и дёргает гварда”.

А чтобы собрать энв, мы в каждом роуте передаем в энв-билдер или в класс-зависимость правильный репо и фильтры. И делов)

А репо 99% у вас абстрактный и конкретные репо наследуются от общего родителя. И там в родителе есть метод find_one(self, **filter_by: dict), а в детях он наследуется. Название замените на своё :)

П.С. почему зависимость, а не мидлвэр - тк в фастапи мидлвэр отрабатывает раньше резолва роутинга. А нам понадобятся параметры пути.

П.С.С. в моем бесплатном открытом курсе на степике есть пример работы с паттерном репозитория и тем, как это все делается лаконично и общо.

А теперь спойлер кода, как пример (но может сделаете еще красивее):

# ваша цель, плюс минус
Depends(
    require_access(
        guard=guard,  # где-то у вас уже заинициализирован гвард с политикой
        env_builder=EnvBuilder(  # напишите сами класс с методом call или можно функцию, смотрите сами, это для примера
            repo=PostRepo,
            filter_by={
                "id": path("post_id"),
            },
            resource_type="post",
            owner_attr="user_id",  # по какому атрибуту поймем кто владелец - для юзера просто айди, для других user_id       
            action="update",
        ),
    )
)

А сам энв билдер напишете сами, потому что я не знаю деталей кода.

Итого, ваша задача написать этот класс, а может даже всего лишь функцию.

Если нужна помощь, то напишите что, постараюсь ответить или накидать на коленке пример класса билдера окружения с теми данными, которые вы мне дали)

[Vladislav2951]

Спасибо за наводку! На днях соберу мысли в кучу и буду пробовать реализовывать :)