
SaaS-продукт со множеством клиентов (арендаторов) рано или поздно сталкивается с выбором: держать раздельные базы на каждого — безопасно, но дорого, или всё сложить в одну базу — дёшево, но страшно. Row-Level Security (RLS) в PostgreSQL позволяет удержать баланс: хранить данные всех арендаторов в одной базе, но технически запретить доступ к «чужим» строкам на уровне СУБД.
Что получает бизнес:
Для RLS нужно явное поле, по которому мы делим строки. Обычно это tenant_id (UUID). Важно, чтобы оно присутствовало во всех таблицах с данными арендатора и участвовало в ключах и связях.
-- Для наглядности — аккаунты-арендаторы
CREATE TABLE tenants (
id uuid PRIMARY KEY,
name text NOT NULL,
created_at timestamptz NOT NULL DEFAULT now()
);
-- Базовая бизнес-таблица, например, проекты
CREATE TABLE projects (
tenant_id uuid NOT NULL,
id bigserial NOT NULL,
name text NOT NULL,
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, id), -- составной ключ: проект уникален внутри арендатора
FOREIGN KEY (tenant_id) REFERENCES tenants(id)
);
-- Задачи внутри проекта
CREATE TABLE tasks (
tenant_id uuid NOT NULL,
id bigserial NOT NULL,
project_id bigint NOT NULL,
title text NOT NULL,
status text NOT NULL CHECK (status IN ('new','in_progress','done')),
created_at timestamptz NOT NULL DEFAULT now(),
PRIMARY KEY (tenant_id, id),
FOREIGN KEY (tenant_id) REFERENCES tenants(id),
-- Внешний ключ тоже включает tenant_id: это критично для целостности в мультиарендности
FOREIGN KEY (tenant_id, project_id) REFERENCES projects(tenant_id, id)
);
-- Индексы под типичные выборки
CREATE INDEX idx_projects_tenant_created_at ON projects (tenant_id, created_at DESC);
CREATE INDEX idx_tasks_tenant_project_id ON tasks (tenant_id, project_id);
Ключевая идея: tenant_id входит во все первичные и внешние ключи. Это исключает «скрещивание» записей между арендаторами на уровне ссылочной целостности, а не только в приложении.
Сначала определим безопасную функцию, которая будет получать tenant_id из параметра сессии. Используем настраиваемый параметр app.tenant_id. Он живёт в пределах транзакции/сессии и надёжно читается в SQL.
-- Функция возвращает текущий tenant_id из параметра сессии
CREATE OR REPLACE FUNCTION app.current_tenant()
RETURNS uuid
LANGUAGE sql
STABLE
AS $$
SELECT current_setting('app.tenant_id', true)::uuid
$$;
Теперь включим RLS и создадим политики для чтения/изменения. Политики используют сравнение tenant_id со значением из функции.
-- Включаем RLS на таблицах
ALTER TABLE projects ENABLE ROW LEVEL SECURITY;
ALTER TABLE tasks ENABLE ROW LEVEL SECURITY;
-- Безопасность по умолчанию: запрещать всё, что не разрешено политиками
ALTER TABLE projects FORCE ROW LEVEL SECURITY;
ALTER TABLE tasks FORCE ROW LEVEL SECURITY;
-- Политики чтения
CREATE POLICY projects_select ON projects
FOR SELECT
USING (tenant_id = app.current_tenant());
CREATE POLICY tasks_select ON tasks
FOR SELECT
USING (tenant_id = app.current_tenant());
-- Политики вставки/обновления/удаления
CREATE POLICY projects_write ON projects
FOR ALL
TO PUBLIC
USING (tenant_id = app.current_tenant())
WITH CHECK (tenant_id = app.current_tenant());
CREATE POLICY tasks_write ON tasks
FOR ALL
TO PUBLIC
USING (tenant_id = app.current_tenant())
WITH CHECK (tenant_id = app.current_tenant());
USING отвечает за строки, которые видит и может редактировать запрос. WITH CHECK — за строки, которые можно создать или во что можно обновить. В обеих проверках мы требуем совпадение tenant_id.
Приложение должно устанавливать параметр app.tenant_id на каждую транзакцию. Делайте это через SET LOCAL — тогда значение живёт только до конца транзакции и не протечёт на следующий запрос в пуле соединений.
Пример на Go с pgxpool: в каждом запросе мы открываем транзакцию, выставляем tenant и выполняем работу.
package main
import (
"context"
"fmt"
"log"
"net/http"
"os"
"github.com/jackc/pgx/v5/pgxpool"
)
func main() {
ctx := context.Background()
dsn := os.Getenv("DATABASE_URL")
pool, err := pgxpool.New(ctx, dsn)
if err != nil { log.Fatal(err) }
defer pool.Close()
http.HandleFunc("/projects", func(w http.ResponseWriter, r *http.Request) {
tenant := r.Header.Get("X-Tenant-ID")
if tenant == "" { http.Error(w, "missing tenant", http.StatusBadRequest); return }
tx, err := pool.BeginTx(ctx, pgxpool.TxOptions{})
if err != nil { http.Error(w, err.Error(), 500); return }
defer tx.Rollback(ctx)
if _, err := tx.Exec(ctx, "SET LOCAL app.tenant_id = $1", tenant); err != nil {
http.Error(w, err.Error(), 500); return
}
rows, err := tx.Query(ctx, "SELECT id, name FROM projects ORDER BY created_at DESC LIMIT 10")
if err != nil { http.Error(w, err.Error(), 500); return }
defer rows.Close()
var out string
for rows.Next() {
var id int64
var name string
if err := rows.Scan(&id, &name); err != nil { http.Error(w, err.Error(), 500); return }
out += fmt.Sprintf("%d\t%s\n", id, name)
}
if err := tx.Commit(ctx); err != nil { http.Error(w, err.Error(), 500); return }
w.Write([]byte(out))
})
log.Println("listening on :8080")
log.Fatal(http.ListenAndServe(":8080", nil))
}
Важно: не используйте SET без LOCAL в пулах соединений (например, PgBouncer в режиме session). Используйте транзакции и SET LOCAL — так безопаснее.
Если у вас Node.js (node-postgres):
const { Pool } = require('pg');
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
async function withTenant(tenantId, fn) {
const client = await pool.connect();
try {
await client.query('BEGIN');
await client.query('SET LOCAL app.tenant_id = $1', [tenantId]);
const res = await fn(client);
await client.query('COMMIT');
return res;
} catch (e) {
await client.query('ROLLBACK');
throw e;
} finally {
client.release();
}
}
// Пример использования
withTenant('a6a8f3e2-5f7b-4c7d-9f0b-2a1e9e1f1f22', async (client) => {
const { rows } = await client.query('SELECT count(*) FROM tasks');
console.log(rows[0]);
});
В мультиарендной схеме важно:
-- Например, хотим уникальные названия проектов внутри арендатора
CREATE UNIQUE INDEX uq_projects_tenant_name ON projects (tenant_id, lower(name));
-- И уникальный внешний идентификатор задачи внутри арендатора
ALTER TABLE tasks ADD COLUMN external_id text;
CREATE UNIQUE INDEX uq_tasks_tenant_external_id ON tasks (tenant_id, lower(external_id));
Если нужен глобально уникальный идентификатор (например, публичный токен), храните его отдельно и не используйте его в связях между таблицами с tenant_id.
RLS сама по себе почти не «стоит» в плане производительности: оптимизатор учитывает условие RLS как дополнительный фильтр. Узкие места обычно в другом:
Пример быстрой выборки последних задач по проекту арендатора:
EXPLAIN ANALYZE
SELECT id, title
FROM tasks
WHERE project_id = 123 -- RLS автоматически добавит tenant_id = app.current_tenant()
ORDER BY id DESC
LIMIT 50;
Если индекс (tenant_id, project_id, id DESC) существует, запрос будет идти по нему без лишних чтений.
-- Роль приложения не должна уметь обходить RLS
ALTER ROLE app_user NOBYPASSRLS;
-- Роль администратора может
ALTER ROLE admin_user BYPASSRLS;
Если делаете админскую функцию, не злоупотребляйте SECURITY DEFINER. Если это необходимо, явно вырубайте/включайте RLS внутри функции и убедитесь, что параметры валидируются.
CREATE OR REPLACE FUNCTION admin.recount_all()
RETURNS void
LANGUAGE plpgsql
SECURITY DEFINER
AS $$
BEGIN
PERFORM set_config('role', 'admin_user', true);
-- Админские действия, где RLS не мешает
-- ...
END;
$$;
Типовой путь из «общей базы без RLS» в «RLS с tenant_id» выглядит так:
ALTER TABLE projects ADD COLUMN tenant_id uuid;
ALTER TABLE tasks ADD COLUMN tenant_id uuid;
-- Временный триггер на вставку: для демонстрации — возьмём из сессии
CREATE OR REPLACE FUNCTION app.set_tenant_on_insert()
RETURNS trigger
LANGUAGE plpgsql
AS $$
BEGIN
IF NEW.tenant_id IS NULL THEN
NEW.tenant_id := current_setting('app.tenant_id', true)::uuid;
END IF;
RETURN NEW;
END;$$;
CREATE TRIGGER trg_projects_tenant BEFORE INSERT ON projects
FOR EACH ROW EXECUTE FUNCTION app.set_tenant_on_insert();
CREATE TRIGGER trg_tasks_tenant BEFORE INSERT ON tasks
FOR EACH ROW EXECUTE FUNCTION app.set_tenant_on_insert();
-- Пример порционного обновления: берём по диапазону id
-- Повторяйте, меняя границы, пока tenant_id не заполнен у всех
UPDATE projects p
SET tenant_id = t.id
FROM tenants t
WHERE p.tenant_id IS NULL
AND p.owner_tenant_ref = t.legacy_owner_ref -- используйте ваш способ связки
AND p.id BETWEEN 100001 AND 120000;
После бекфила сделайте NOT NULL, добавьте составные ключи и индексы CONCURRENTLY.
ALTER TABLE projects ALTER COLUMN tenant_id SET NOT NULL;
ALTER TABLE tasks ALTER COLUMN tenant_id SET NOT NULL;
-- Перестраиваем ключи и связи без долгих блокировок
ALTER TABLE projects DROP CONSTRAINT projects_pkey;
ALTER TABLE projects ADD PRIMARY KEY USING INDEX
(SELECT indexname FROM pg_indexes WHERE tablename='projects' LIMIT 0); -- см. примечание ниже
Примечание: на бою готовьте создание новых индексов заранее через CREATE UNIQUE INDEX CONCURRENTLY, затем переключайте ключи на них в короткую блокировку. Конкретные команды зависят от вашей схемы, поэтому обязательно отрепетируйте на копии базы.
CREATE VIEW app.safe_projects WITH (security_barrier=true) AS
SELECT id, name, created_at FROM projects WHERE tenant_id = app.current_tenant();
Итог: RLS — это простая, но мощная защита на уровне базы. Она дисциплинирует схему, снижает риск утечек из-за человеческого фактора и позволяет экономить на инфраструктуре, не жертвуя скоростью разработки. При грамотной схеме ключей и индексах RLS работает быстро и прозрачно для кода — вы получаете мультиарендный продукт без отдельных баз и лишних серверов.