Здравствуйте!

Статья о создании супергибкой системы проверки прав к ресурсам на основе клэймов и политик. В свое время был вдохновлен реализацией Claims‑based authorization in ASP.NET Core, но нужно было на Node.js (а теоретически на любую платформу, где используется PostgreSQL). Поэтому решил написать функцию проверки прав на PL/pgSQL. Если интересно, го.

Предисловие

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

Пользователь должен быть авторизован через Facebook, старше 18 лет, но моложе 50. А также должен состоять в сообществе любителей машин Феррари.
Пользователь должен выполнить какое‑то задание, после этого откроется доступ к ресурсу.
Все совершеннолетние, не проживающие в Москве.

Сценариев можно придумать очень много под разные бизнес‑задачи.

Написанная мною система, к сожалению, свет не увидела (по причине финансирования проекта) и пока не была протестирована в продакшене.

Коллеги, с которыми обсуждал эту систему, и я были довольны — система идеально вписывалась в наши требования. Хочу услышать мнения профессионалов и получить обратную связь.

Термины

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

Тип ресурса — тип проверяемых данных. Например, посты блога, заказы.

Ресурс (сущность, entity, resource) — то, с чем пользователь может взаимодействовать. Это может быть пост в блоге, заказ на сайте. В общем, это то, к чему будем проверять права. Не путать с типом ресурса. Ресурс — это конкретная сущность (как правило — то, что имеет id), а тип объединяет все сущности.

Возможность ресурса (ability) — действие, которое пользователь может совершить над ресурсом. У одного ресурса может быть много возможностей. Например, ресурс — пост в блоге. Возможности ресурса — чтение, правка, удаление.

Клэйм (кляйм, claim) — конкретная информация о пользователе. Например, возраст, в каких группах состоит и т. д. На момент проверки прав пользователь предоставляет все свои клэймы, и система на основе этих клэймов проверяет, имеет ли пользователь доступ к тому или иному ресурсу.

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

Принцип работы

Представим, что форма управления доступа к посту выглядит так.

Форма редактирования доступа для тех, кто может редактировать пост

Поясню: править пост может администратор или редактор старше 18 лет, проживающий в Москве или Санкт‑Петербурге. Или Иванов И.И. с Id 123. Давайте упростим. Мы используем 4 политики. Введем сокращения:

P1 — Состоит в группе
P2 — Возраст старше
P3 — Город проживания
P4 — Пользователь

Тогда данные нашей формы можно представить выражением:

(
  P1 = “Администратор”
  ИЛИ 
  (
    P1 = “Редактор” 
    И 
    P2 >= 18 
    И 
    (
      P3 = “Москва” 
      ИЛИ 
      P3 = “Санкт-Петербург”
    )
  ) 
  ИЛИ 
  P4 = 123
)

Немного напоминает SQL выражение WHERE, не правда ли? И наша задача сводится к тому, чтобы эту форму привести в готовое валидное SQL WHERE выражение и в момент проверки доступа на правку этого поста вызвать его.

Начнём погружение

Создадим таблицу с постами и заполним её данными.

CREATE TABLE IF NOT EXISTS post (
   id serial primary key,
   title text not null
);


INSERT INTO post(title) VALUES ('Post 1'), ('Post 2');

Проверка прав идёт в процессе выборки данных. Когда мы получаем посты, мы делаем запрос.

Так мы выбираем все посты

SELECT p.*
FROM post p;

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

Код системы

CREATE TYPE group_condition AS enum ('OR', 'AND');

CREATE TABLE IF NOT EXISTS policy (
    id serial primary key,
    name varchar,
    resolver_code  varchar(255) not null
);

CREATE TABLE IF NOT EXISTS claim (
    id serial primary key,
    code varchar,
    name varchar,
    replaceable_by_user boolean default true
);

CREATE TABLE IF NOT EXISTS policy_claim (
    policy_id integer not null
        references policy on delete restrict,
    claim_id integer not null references claim on delete restrict,
    where_condition text default '%1$s = %2$s'::text not null,
    primary key (policy_id, claim_id)
);


CREATE TABLE IF NOT EXISTS post_group_policy (
    id serial primary key,
    type group_condition       not null,
    parent_group integer
        constraint post_group_policy_post_group_policy_id_fk
            references post_group_policy
            on delete cascade,
    negative boolean default false not null
);

CREATE TABLE IF NOT EXISTS post_ability (
    code varchar not null primary key,
    name varchar not null
);

CREATE TABLE IF NOT EXISTS post_policy (
    id serial primary key,
    entity_id int not null constraint post_policy_post_id_fk
            references post on delete cascade,
    policy_id integer not null references policy on delete restrict,
    group_id integer references post_group_policy on delete cascade,
    ability varchar not null references post_ability on delete restrict,
    negative  boolean default false not null
);

CREATE TABLE IF NOT EXISTS post_policy_claim (
    entity_policy_id integer not null
        constraint post_policy_claim_post_policy_id_fkey
            references post_policy
            on delete cascade,
    claim_id integer not null
        references claim
            on delete restrict,
    value varchar not null,
    primary key (entity_policy_id, claim_id, value)
);

CREATE TABLE IF NOT EXISTS post_indexed_rule (
    entity_id int not null,
    condition text not null,
    ability varchar not null references post_ability on delete cascade,
    used_claims character varying[] not null,
    constraint post_indexed_rule_pk
        primary key (ability, entity_id)
);

/**
  Функция чтобы перевернуть условие, если нужно
 */
DROP FUNCTION IF EXISTS __process_policy_negative;
CREATE FUNCTION __process_policy_negative(_condition text, _is_negative boolean)
    RETURNS text
AS $$
DECLARE
    _format text;
BEGIN
    IF _is_negative THEN
        _format = 'NOT(%s)';
    ELSE
        _format = '%s';
    END IF;
    RETURN format(_format, _condition);
END
$$ LANGUAGE plpgsql;

/**
  Обработка условий для группы
  Функция рекурсивная. Ей на вход приходит корневая группа, и вызывается вниз по дереву для каждой группы
 */
DROP FUNCTION IF EXISTS __process_policy_group_condition;
CREATE FUNCTION __process_policy_group_condition(_entity_policy_group record, _query text)
    RETURNS text
AS $$
DECLARE
    _group_policies record;
    _format text;
BEGIN
    -- склейм условия группы с родительской группой (или ни с чем если ф-ия вызывается первый раз)
    _query = format('%s%s', _query, _entity_policy_group.group_condition);

    -- найдём все дочерние группы текущей группы
    BEGIN FOR _group_policies IN
        SELECT *
        FROM policy_group_condition p
        WHERE p.parent_group = _entity_policy_group.group_id
        LOOP

            -- если группа с отрицанием, то нужно перевернуть условие дочерней группы
            IF _entity_policy_group.negative THEN
                _format = '%s %s NOT(';
            ELSE
                _format = '%s %s (';
            end if;
            _query := format(_format, _query, _entity_policy_group.type);
            -- обработаем дочернюю группу рекурсивно
            SELECT __process_policy_group_condition(_group_policies, _query) INTO _query;
            _query := _query || ')';
        END LOOP;
    END;
    return _query;
END;
$$ LANGUAGE plpgsql;

DROP FUNCTION IF EXISTS __process_user_claim_value;
CREATE FUNCTION __process_user_claim_value(_user_claim_value_jsonb jsonb)
    RETURNS text
AS $$
DECLARE
    _user_claim_value text;
    _user_claim_value_type text;
    _user_claim_value_array_item_jsonb jsonb;
    _user_claim_value_array_item_type text;
    _user_claim_value_array_int_result int[] DEFAULT ARRAY[]::int[];
    _user_claim_value_array_text_result text[] DEFAULT ARRAY[]::text[];
BEGIN
    -- текстовое значение кляйма
    _user_claim_value := _user_claim_value_jsonb ->> 0;

    -- если пользователь не передал кляйм, значит будем проверять на пустоту (т.е. политика должна требовать пустой кляйм)
    IF _user_claim_value IS NULL THEN
        _user_claim_value = format('%L', '');
    ELSE
        -- получаем тип переданого кляйма
        _user_claim_value_type = jsonb_typeof(_user_claim_value_jsonb);
        CASE

            -- если это массив
            WHEN _user_claim_value_type = 'array' THEN

                -- если не пустой, тогда превратим jsonb массив в строку, соединенную запятымы ([1,2,3,4] => '1,2,3,4')
                IF jsonb_array_length(_user_claim_value_jsonb) > 0 THEN
                    _user_claim_value := format('%L', '');

                    -- тут надо правильно преобразовать jsonb массив в строку для поиска.
                    -- с интами все просто [1,2,3] => 1,2,3
                    -- а стринги нужно из ["s1", "s2"] преобразовать в 's1','s2' (поменять кавычки)
                    -- для этого мы ходим по каждому элементу массива
                    -- тут мы используем 2 массива: числовой (_user_claim_value_array_int_result) и текстовой (_user_claim_value_array_text_result)
                    -- в зависимости от типа значения, мы помещаем его в нужный массив
                    -- потом мы массив, в котором есть значения, конкатим по запятой и присваиваем результату
                    FOR _user_claim_value_array_item_jsonb IN SELECT a FROM jsonb_array_elements(_user_claim_value_jsonb) as a
                        LOOP
                            -- берем его тип
                            _user_claim_value_array_item_type = jsonb_typeof(_user_claim_value_array_item_jsonb );
                            CASE
                                -- если это инт, то пихаем как есть в числовой массив
                                WHEN _user_claim_value_array_item_type = 'number' THEN
                                    SELECT array_append(_user_claim_value_array_int_result, _user_claim_value_array_item_jsonb::int) INTO _user_claim_value_array_int_result;

                                -- если это строка, то экстрактим её значение и пихаем в текстовый массив
                                WHEN _user_claim_value_array_item_type = 'string' THEN
                                    SELECT array_append(_user_claim_value_array_text_result, format('%L', _user_claim_value_array_item_jsonb ->> 0)) INTO _user_claim_value_array_text_result;
                                END CASE;
                        END LOOP;

                    -- если есть значения в текстовом массиве
                    IF array_length(_user_claim_value_array_text_result, 1) > 0 THEN
                        SELECT INTO _user_claim_value array_to_string(_user_claim_value_array_text_result, ',');

                        -- если есть значения в числовом массиве
                    ELSIF array_length(_user_claim_value_array_int_result, 1) > 0 THEN
                        SELECT INTO _user_claim_value array_to_string(_user_claim_value_array_int_result, ',');
                    END IF;

                ELSE
                    _user_claim_value = 'NULL';
                END IF;
            ELSE

                -- если это скалярное выражение, то просто обернем его в кавычки
                _user_claim_value = format('%L', _user_claim_value);
            END CASE;
    END IF;

    RETURN _user_claim_value;
END;
$$ LANGUAGE plpgsql;


DROP FUNCTION IF EXISTS index_entity_policies;
CREATE FUNCTION index_entity_policies(_table_prefix varchar, _entity_id anyelement, _ability anycompatible)
    RETURNS void
AS $$
DECLARE
    _policy_group_condition record;
    _check_query text;
    _check_where text default '';
    _used_claims varchar[] default ARRAY[]::varchar[];
BEGIN
    -- сохраним во временную таблицу все сохраненные политики для этой сущности
    CREATE TEMPORARY TABLE IF NOT EXISTS entity_policies (
                                                             id int,
                                                             policy_id int,
                                                             group_id int,
                                                             negative boolean
    );

    TRUNCATE TABLE entity_policies;
    EXECUTE format('INSERT INTO entity_policies ' ||
                   'SELECT id, policy_id, group_id, negative FROM %I WHERE entity_id = $1 AND ability = $2;', _table_prefix || '_policy')
        USING  _entity_id, _ability;

    -- если нет политик или возможностей
    IF NOT EXISTS(SELECT 1 from entity_policies) THEN
        RETURN;
    END IF;

    -- сохраним во временную таблицу все кляймы, которые требуются. подцепим к ним информацию о группе, в которой находятся политики
    CREATE TEMPORARY TABLE IF NOT EXISTS required_policy_claims(
                                                                   id int,
                                                                   policy_id int,
                                                                   claim_id int,
                                                                   code text,
                                                                   value text,
                                                                   where_condition text,
                                                                   group_id int,
                                                                   parent_group int,
                                                                   type group_condition,
                                                                   negative boolean,
                                                                   entity_policy_negative boolean,
                                                                   replaceable_by_user_claim boolean
    );
    TRUNCATE TABLE required_policy_claims;
    EXECUTE format(
                                    'INSERT INTO required_policy_claims ' ||
                                    'SELECT ep.id, ep.policy_id, stpc.claim_id, c.code, stpc.value, pc.where_condition, ep.group_id, stgp.parent_group, stgp.type, stgp.negative, ep.negative as entity_policy_negative, c.replaceable_by_user replaceable_by_user_claim ' ||
                                    'FROM %I stpc ' ||
                                    'INNER JOIN entity_policies ep on stpc.entity_policy_id = ep.id ' ||
                                    'INNER JOIN claim c on c.id = stpc.claim_id ' ||
                                    'INNER JOIN %I stgp on stgp.id = ep.group_id ' ||
                                    'INNER JOIN policy_claim pc on ep.policy_id = pc.policy_id AND stpc.claim_id = pc.claim_id',
                                    _table_prefix || '_policy_claim', _table_prefix || '_group_policy'
        );

    -- в эту временную таблицу мы складываем уже сгруппированные политики по группам
    CREATE TEMPORARY TABLE IF NOT EXISTS policy_group_condition (
                                                                    group_id int,
                                                                    parent_group int,
                                                                    group_condition text,
                                                                    type text,
                                                                    negative boolean
    );
    TRUNCATE TABLE policy_group_condition;
    INSERT INTO policy_group_condition
    WITH policits_with_cond AS (
        SELECT rpc.id,
               -- тут мы объединяем кляймы одной политики по условию AND
               format('(%s)', array_to_string(array_agg(format(__process_policy_negative(rpc.where_condition, rpc.entity_policy_negative), quote_literal(rpc.value), format('[#%s#]', rpc.code))), ') AND (')) as condition,
               rpc.group_id,
               rpc.parent_group,
               rpc.type,
               rpc.negative
        FROM required_policy_claims rpc
        GROUP BY rpc.id, rpc.type, rpc.group_id, rpc.parent_group, rpc.type, rpc.negative
        ORDER BY rpc.id
    ), policits_by_groups AS (
        SELECT
            p.group_id,
            p.parent_group,
            -- объединяем политики в группе по условию, прописанному в группе
            format('(%s)', array_to_string(array_agg(__process_policy_negative(p.condition, p.negative)), format(') %s (', p.type))) as group_condition,
            p.type,
            p.negative
        FROM policits_with_cond p
        GROUP BY p.group_id, p.type, p.parent_group, p.negative
        ORDER BY p.group_id
    ) SELECT * FROM policits_by_groups;

    -- начем рекурсивно проходить по группам. начинаем с корневой группы
    BEGIN FOR _policy_group_condition IN
        SELECT * FROM policy_group_condition p WHERE p.parent_group IS NULL
        LOOP
            SELECT __process_policy_group_condition(_policy_group_condition, _check_where) INTO _check_where;
        END LOOP;
    END;

    SELECT INTO _used_claims ARRAY(SELECT DISTINCT code FROM required_policy_claims WHERE replaceable_by_user_claim = true);

    -- иготовый запрос, который будем проверять.
    _check_query := format('SELECT true WHERE (%s)', _check_where);

    EXECUTE format('INSERT INTO %I(entity_id, ability, condition, used_claims) ' ||
                   'VALUES ($1, $2, $3, $4) ' ||
                   'ON CONFLICT (entity_id, ability) DO UPDATE ' ||
                   'SET condition = $3, used_claims = $4', _table_prefix || '_indexed_rule')
        USING _entity_id, _ability, _check_where, _used_claims;


END
$$ LANGUAGE plpgsql;

DROP FUNCTION IF EXISTS check_user_rights;
CREATE FUNCTION check_user_rights(_table_prefix varchar, _entity_id anyelement, _ability anycompatible, _user_claims jsonb)
    RETURNS boolean
AS $$
DECLARE
    _where_condition record;
    _used_claim varchar;
    _result_condition text;
    _check_query boolean;
BEGIN

    EXECUTE format('SELECT condition, used_claims ' ||
                   'FROM %I WHERE entity_id = $1 AND ability = $2;', _table_prefix || '_indexed_rule')
        USING _entity_id, _ability INTO _where_condition ;

    _result_condition = _where_condition.condition;

    IF _where_condition IS NULL THEN
        RETURN NULL;
    END IF;

    FOREACH _used_claim IN ARRAY _where_condition.used_claims
        LOOP
            SELECT INTO _result_condition replace(_result_condition, format('[#%s#]', _used_claim), COALESCE(_user_claims ->> _used_claim, quote_literal('')));
        END LOOP;

    EXECUTE format('SELECT true WHERE (%s)', _result_condition) INTO _check_query;
    RETURN _check_query;
END
$$ LANGUAGE plpgsql;


DROP FUNCTION IF EXISTS __process_user_claim_values;
CREATE FUNCTION __process_user_claim_values(_user_claims jsonb)
    RETURNS jsonb
AS $$
DECLARE
    _json_el record;
    _jsonb_result jsonb default '{}'::jsonb;
BEGIN
    BEGIN FOR _json_el IN
        select key, value from jsonb_each(_user_claims)
        LOOP
            SELECT INTO _jsonb_result jsonb_set(_jsonb_result, ARRAY[_json_el.key], to_jsonb(__process_user_claim_value(_json_el.value)));
        END LOOP;
    END;
    RETURN _jsonb_result;
END;
$$ LANGUAGE plpgsql;

DROP FUNCTION IF EXISTS check_index_rules CASCADE;
CREATE FUNCTION check_index_rules() RETURNS trigger AS $$
DECLARE
    _rows_count int;
BEGIN
    EXECUTE format('SELECT count(*) FROM %I ' ||
                   'WHERE entity_id = $1 AND ability = $2', TG_ARGV[0] || '_policy')
        USING old.entity_id, old.ability INTO _rows_count ;

    IF _rows_count = 0 THEN
        EXECUTE format('DELETE FROM %I WHERE entity_id = $1 AND ability = $2', TG_ARGV[0] || '_indexed_rule')
            USING old.entity_id, old.ability;

    END IF;
    RETURN new;
END;
$$ LANGUAGE plpgsql;

-- тригер, который следит за правилами. Если будет удалено последнее правило, то удалится и проиндексированное правило
CREATE TRIGGER post_clear_index AFTER DELETE ON post_policy
	FOR EACH ROW EXECUTE FUNCTION check_index_rules('post');

Изменим запрос на выборку под наши нужды.

Запрос для выборки постов с проверкой прав

SELECT p.*
FROM
   post p,
   LATERAL (SELECT /* (1) */ __process_user_claim_values(/* (2) */ '{"age": 24, "location": "Москва", "userId": 123}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
   'post', /* (3) */
   p.id, /* (4) */
   'edit', /* (5) */
   l.processed_user_claim /* (6) */
);

Тут у нас 24-летний пользователь из Москвы пытается получить список постов, которые он может редактировать. Функция check_user_rights должна проверить, может ли наш пользователь редактировать эти посты.

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

Давайте разберём эту выборку.

Разбор выборки данных

/* (1) */ преобразование клэймов

В выражении FROM мы преобразовываем клэймы пользователя в алиас l. Это делается тут, чтобы преобразование прошло один раз, а не при каждой проверке в WHERE.

Что делает функция __process_user_claim_values? Она преобразует значения jsonb в значения, которые можно использовать в условии WHERE. Поясню на примере.

Преобразование клэймов в SQL пригодные литералы

SELECT __process_user_claim_values('{
   "intValue": 12345,
   "stringValue": "abcde",
   "stringArrayValue": ["a", "b", "c"],
   "intArrayValue": [1,2,3,4,5]
}'::jsonb);

Результатом будет немного измененный jsonb.

{
   "intValue": "'12345'",
   "stringValue": "'abcde'",
   "intArrayValue": "1,2,3,4,5",
   "stringArrayValue": "'a','b','c'"
}

intValue, stringValue - значения обернулись в кавычки.
intArrayValue - массив преобразовался в строку, где значения разделены запятой.
stringArrayValue - массив также преобразовался в строку, где значения разделены запятой, но каждое значение обернуто в кавычки.

Теперь эти значения можно вставлять в WHERE выражения. Например, intArrayValue и stringArrayValue можно использовать в выражении IN.

SELECT … FROM … WHERE p.id IN (1,2,3,4,5).

/* (2) */ откуда берутся клэймы

На практике информация о пользователе может храниться в БД (принадлежность к группе, возраст и т.д.). Также часть информации может храниться, например, в Redis.

Наша цель - собрать отовсюду информацию о пользователе и передать ее в функцию __process_user_claim_values. И дальше уже преобразованные клэймы передать в функцию check_user_rights, которая и решит, имеет ли пользователь доступ.

/* (3) */ название типа ресурса

Первым параметром функция check_user_rights принимает название типа ресурса. Этим параметром мы говорим функции, что проверять будем посты. К смыслу префикса мы еще вернемся впоследствии.

/* (4) */ id ресурса

Второй параметр - это id ресурса. Функция будет проверять, имеет ли пользователь доступ к конкретно этому ресурсу.

/* (5) */ возможность

Третий параметр нужен для проверки возможности. У нас указано edit, т.е. мы проверяем, имеет ли пользователь возможность редактировать пост.

/* (6) */ преобразованные клэймы

И последним параметром в функцию передаем клэймы, которые мы преобразовали выше. Если бы мы не делали преобразование, а сразу бы передали в функцию необработанные клэймы, то производительность заметно бы пострадала.

Резюме

Итак, этим запросом мы говорим:

"Выбери все посты, которые 24-летний пользователь из Москвы с id 123 может редактировать".

Но что же происходит в функции check_user_rights? Прежде чем продолжить, предлагаю изучить структуру БД, и что надо сделать до непосредственной проверки.

Структура базы данных

Нам нужно создать 3 общие таблицы для всех типов ресурсов и 5 таблиц на каждый тип ресурса.

Общие таблицы:

policy - политики, доступные для выбора.
claim - клэймы, доступные для создания политик.
policy_claim - клэймы, связанные с политиками. Тут мы указываем политикам, какие клэймы они используют и как.

Для проверки прав к ресурсу создаётся 5 таблиц с префиксом в названии. Префикс для таблиц - название типа ресурса. В примере ниже префикс post. Для заказов мы сделаем префикс - order и т.д. Важно понять, что для каждого типа ресурса, будет свой набор таблиц. Сами таблицы:

post_ability - перечислены все возможности, которые использует ресурс. Например, ресурс пост может иметь возможности: просмотр, правка, удаление.
post_group_policy - группировка политик ресурса. Политики группируются для сравнения друг с другом по правилу И, ИЛИ, И НЕ, ИЛИ НЕ. Например, конкретному посту можно сказать: возраст больше 18 И НЕ живёт в Москве или возраст больше 18 ИЛИ живёт в Москве.
post_policy - связь ресурса с политикой. Тут указывается, какие политики применяются к конкретному ресурсу.
post_policy_claim - указываются конкретные клэймы к политикам. Например, для поста указана политика "Возраст старше". Эта политика требует клэйм "возраст", в этой таблице и будет указан требуемый возраст.
post_indexed_rule - правило для проверки.

Добавим правило для поста

Давайте вручную добавим правило, проговаривая все этапы.

Сначала к посту добавим политику "состоит в группе" со значением "администратор". Представим, что группа “администратор” имеет Id = 1.

Добавим политику в таблицу policy.

INSERT INTO policy (name, resolver_code) VALUES('Состоит в группе', 'inGroup');

Добавили название и код для политики (мало ли для чего он может понадобиться). Этой политики присвоился Id = 1. Эту политику можно будет использовать для любых типов ресурсов.

Добавим клэйм в таблицу claim.

INSERT INTO claim(name, code) VALUES ('Группа пользователя', 'group');

Также этот клэйм будет доступен для всех видов ресурсов. Клэйму присвоился Id = 1.

Связать добавленный клэйм с политикой.

INSERT INTO policy_claim (policy_id, claim_id, where_condition) 
VALUES(1, 1, '%1$s::int IN (%2$s)');

Значение из колонки where_condition будет подставлено в финальное WHERE выражение. Это значение будет подставлено в функцию format.

В позицию %1$s - будет подставлен требуемый клэйм для поста этой политики (в нашем случае 1, т.к. мы представляем, что это Id группы “администратор”).

В позицию %2$s - будет подставлено значение клэймов, которые передал пользователь. Т.к. пользователь может состоять в нескольких группах, мы используем выражение IN.

Представим, что политика требует, чтобы пользователь был в группе с Id 1, а пользователь состоит в группах 1,2,3. Тогда результатом where_condition будет 1::int in (1,2,3).

Добавим возможность правки для поста.

INSERT INTO post_ability (code, name) VALUES ('edit', 'Правка');

Этим запросом мы создаем новую возможность для постов.

Добавим правило для поста.

Будем назначать права для поста с Id 1. Первый запрос создает группу политик поста.

INSERT INTO post_group_policy (type, parent_group, negative) 
VALUES ('OR', null, false);

Запрос создал группу с Id 1. parent_group у нас null (значит группа правил находится на верхнем уровне). parent_group ссылается на id этой же таблицы. Если он задан, то эта группа правил является дочерней для той, на которую ссылаемся. Все дочерние правила сравниваются между собой по условию из колонки type.

type может быть AND или OR. Таблица с пояснениями ниже.

negative - отрицать ли условия правил.

Таблица сравнения групп правил. P1, P2, P3 - группа дочерних правил. Группа может включать другие группы, которые теоретически могут быть неограниченно вложенными.

negative / condition

OR

AND

true

Любое из дочерних правил должно не удовлетворять условию

WHERE NOT(P1) OR NOT(P2) OR NOT(P3)

Все дочерние правила не должны удовлетворять условию

WHERE NOT(P1) AND NOT(P2) AND NOT(P3)

false

Любое из дочерних правил должно удовлетворять условию

WHERE P1 OR P2 OR P3

Все дочерние правила должны удовлетворять условию

WHERE P1 AND P2 AND P3

Далее

INSERT INTO post_policy(policy_id, entity_id, group_id, negative, ability) 
VALUES (1,1,1, false, 'edit');

Этим запросом мы назначили посту правило на редактирования. Следующий запрос с установкой требуемого клэйма.

Назначаем клэйм.

INSERT INTO post_policy_claim(entity_policy_id, claim_id, value) 
VALUES (1, 1, 1);

В колонку entity_policy_id добавляется Id, созданный в результате предыдущего запроса.

claim_id - клэйм, добавленный в таблицу claims. Клэйм с Id 1 - это "группа пользователя".

value - клэйм требует, чтобы у пользователя была назначена группа 1.

Все данные для правила мы назначили.

Проиндексируем изменения.

SELECT index_entity_policies(/* _table_prefix */ 'post', /* _entity_id */ 1, /* _ability */ 'edit');

Описание параметров:

_table_prefix - префикс для таблиц, а префикс - это тип ресурса. В запросе мы индексируем посты.
_entity_id - id ресурса (сущности). В нашем случае это пост с id = 1.
_ability - возможность, которую индексируем. Т.е. редактирование поста.

В итоге запрос выглядит так: проиндексируй правила на редактирование поста с id = 1.

Результатом этого запроса будет составление WHERE выражения в таблицу post_indexed_rule

(('1'::int in ([#group#]))).

В запрос вставлены плэйсхолдеры [#placeholder_name#]. В момент проверки правила эти плейсхолдеры будут заменены клэймами пользователя.

Также в таблице есть колонка used_claims. В ней хранится массив клэймов, которые используются в WHERE выражении в качестве плэйсхолдера. Колонка используется просто для ускорения замены клэймов в момент проверки прав.

Проверим выборку.

SELECT p.id, p.*
FROM
   post p,
   LATERAL (SELECT __process_user_claim_values('{"group": [1,2]}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
   'post',
   p.id,
   'edit',
   l.processed_user_claim
);

Как мы видим, в выборке одна запись, которая удовлетворяет нашему условию. Пользователь состоит в группе 1.

Представим, что пользователя нет в этой группе.

SELECT p.id, p.*
FROM
   post p,
   LATERAL (SELECT __process_user_claim_values('{"group": [2]}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
   'post',
   p.id,
   'edit',
   l.processed_user_claim
);

А тут уже пусто. Клэймы не удовлетворяют требуемому правилу.

Добавляем остальные правила

Не буду повторяться, остановлюсь только на вещах, которые не описал выше.

Добавление остальных правил

Добавим политики

INSERT INTO policy (name, resolver_code)
VALUES ('Возраст старше', 'age_older'), 
       ('Город проживания', 'location'), 
       ('ID пользователя', 'user_id');

Добавим клэймы.

INSERT INTO claim (code, name)
VALUES ('age', 'Возраст'), 
       ('location', 'Город проживания'), 
       ('user_id', 'ID пользователя');


INSERT INTO policy_claim (policy_id, claim_id, where_condition)
VALUES (2, 2, '%1$s::int <= %2$s::int'), 
       (3, 3, '%1$s = %2$s'), 
       (4, 4, '%1$s = %2$s');

Добавим правила.

-- добавление групп
INSERT INTO post_group_policy (type, parent_group) VALUES ('AND', 1); -- Добавлена группа первого уровня с ID: 2
INSERT INTO post_group_policy (type, parent_group) VALUES ('OR', 2); -- Добавлена группа второго уровня с ID: 3


-- добавление правила
INSERT INTO post_policy (entity_id, policy_id, group_id, ability)
VALUES (1 /* ID поста */, 1 /* Политика "Состоит в группе" */, 2 /* группа первого уровня */ , 'edit'); -- Добавлено правило с ID: 2


INSERT INTO post_policy_claim (entity_policy_id, claim_id, value)
VALUES (2 /* правило, добавленное в пред. запросе */, 1 /* клэйм "Состоит в группе" */, 2 /* ID группы */);


INSERT INTO post_policy (entity_id, policy_id, group_id, ability)
VALUES (1 /* ID поста */, 2 /* Политика "Возраст старше" */, 2 /* группа первого уровня */, 'edit'); -- Добавлено правило с ID: 3


INSERT INTO post_policy_claim (entity_policy_id, claim_id, value)
VALUES (3 /* правило, добавленное в пред. запросе */, 2 /* клэйм "Возраст" */, 18 /* Возраст */);


INSERT INTO post_policy (entity_id, policy_id, group_id, ability)
VALUES (1 /* ID поста */, 3 /* политика "Город проживания" */, 3 /* группа второго уровня */, 'edit'); -- Добавлено правило с ID: 4


INSERT INTO post_policy_claim (entity_policy_id, claim_id, value)
VALUES (4 /* правило, добавленное в пред. запросе */, 3 /* клэйм "Город проживания" */, 'Москва');


INSERT INTO post_policy (entity_id, policy_id, group_id, ability)
VALUES (1 /* ID поста */, 3 /* политика "Город проживания" */, 3 /* группа второго уровня */, 'edit'); -- Добавлено правило с ID: 5


INSERT INTO post_policy_claim (entity_policy_id, claim_id, value)
VALUES (5 /* правило, добавленное в пред. запросе */, 3 /* клэйм "Город проживания" */, 'Санкт-Петербург');


INSERT INTO post_policy (entity_id, policy_id, group_id, ability)
VALUES (1 /* ID поста */, 4 /* политика "ID пользователя" */, 1 /* корневая группа */, 'edit'); -- Добавлено правило с ID: 6


INSERT INTO post_policy_claim (entity_policy_id, claim_id, value)
VALUES (6 /* правило, добавленное в пред. запросе */, 4 /* клэйм "ID пользователя" */, 123);


-- проиндексируем изменения
SELECT index_entity_policies('post', 1, 'edit');

Получилось такое WHERE выражение

(
  ('1'::int in ([#group#]))
) OR (
  ('123' = [#user_id#])
) OR (
  (
    ('2'::int in ([#group#]))
  ) AND (
    ('18'::int < [#age#]::int)
  ) AND (
    (
      ('Москва' = [#location#])
    ) OR (
      ('Санкт-Петербург' = [#location#])
    )
  )
).

Проверим выборку. Предоставляя разные клйэмы.

Проверка 1. Клэймы не удовлетворяют условию

Пользователь состоит в группе 2, 17 лет, живет в Москве. ID = 124

SELECT p.id, p.*
FROM
  post p,
  LATERAL (SELECT __process_user_claim_values('{"group": [2], "age": 17, "location": "Москва", "user_id": 124}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
          'post',
          p.id,
          'edit',
          l.processed_user_claim
       );

Ничего нет. Клэймы не удовлетворяют условию.

Проверка 2. Клэймы удовлетворяют условию. Пользователь состоит в группе 2 (Редактор), ему 25 лет и живет в Москве

Пользователь состоит в группе 2, 25 лет, живет в Москве. ID = 124

SELECT p.id, p.*
FROM
  post p,
  LATERAL (SELECT __process_user_claim_values('{"group": [2], "age": 25, "location": "Москва", "user_id": 124}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
          'post',
          p.id,
          'edit',
          l.processed_user_claim
       );

Проверка 3. Клэймы удовлетворяют условию. ID пользователя = 123

Пользователь состоит в группе 2, 17 лет, живет в Москве. ID = 123

SELECT p.id, p.*
FROM
  post p,
  LATERAL (SELECT __process_user_claim_values('{"group": [2], "age": 17, "location": "Москва", "user_id": 123}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
          'post',
          p.id,
          'edit',
          l.processed_user_claim
       );

Пользователю 17 лет и проверку 2 он уже не пройдет, зато проходит проверку политики ID Пользователя = 123

Проверка 4. Клэймы удовлетворяют условию. Состоит в группе 1

Пользователь состоит в группе 1 и 2, 17 лет, живет в Москве. ID = 124

SELECT p.id, p.*
FROM
  post p,
  LATERAL (SELECT __process_user_claim_values('{"group": [1, 2], "age": 17, "location": "Москва", "user_id": 124}'::jsonb) as processed_user_claim) l
WHERE check_user_rights(
          'post',
          p.id,
          'edit',
          l.processed_user_claim
       );

Проверки 2 и 3 уже не пройдут, зато эти клэймы удовлетворяют политике Состоит в группе = 1

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

Целостность данных

Целостность поддерживается внешними ключами. Единственное, что не поддалось ключам, это таблица post_indexed_rule. Для поддержки актуальности написал триггер на удаление записей из post_policy. При удалении последнего правила удаляется запись.

Резюме

Система получилась очень интересной. Я приобрел ценный опыт. На написание этой системы я потратил около полутора недель с учётом того, что никогда не писал на PL/PgSql.

Что мне нравится:

Гибкость правил.
Скорость. Правила проверяются быстро. Проверил на выборке в 150000 записей. Отработала за 600 мс. Пока для наших задач это выше крыши, но хотелось бы почитать ваши мнения по этому поводу.
Можно использовать на любой платформе, языке. Главное, чтобы PostgreSQL стояла.

Что не нравится:

Для каждого типа ресурса нужно создавать 5 таблиц. Если типов много, то база разрастается огромным количеством таблиц. Не думаю, что это большая проблема, ведь их можно перенести в отдельную схему. Думаю, там им и место.
Как правило, для всех постов правила будут одинаковые, и странно хранить для всех постов одни и те же правила.
Сбор клеймов.

Если будет интересно, представлю клиентский код на typescript с использованием typeorm для работы с системой.

Как вам моя разработка? Буду рад обратной связи. Спасибо за внимание!

Claim-based авторизация на PL/pgSQL