Supabase に Row Level Security を設定する

背景

マルチテナント構成のシステムにて、APIサーバのバックエンドにsupabaseを利用する際、Row Level Security (以下RLS)を設定しておく必要があります。

各テーブルが team_id というフィールドを持っており、アクセスしたユーザーが所属する team_idと一致しているレコードのみにアクセスできるようにする、という使い方です。

このRLSの設定にて、躓いたところと解決法をメモしておきます。

ログインロールについて

supabaseでは、デフォルトで用意されている postgres ロールを利用してDBにつなぐ方法が想定されています。

この postgresロールには問題があって、ロールの設定を確認すると、User bypasses every row level serucirty policy がONになっています。

これは各テーブルに RLSを設定したとしても、postgresロールでDBに接続している限りではRLSを無視してデータにアクセスできてしまうことを示します。

そして、その設定をGUI上で変更することはできないようです。

そのため、postgresではない別のロールを作成し、そのロールでDBにログインするようにします。

RLSを設定するSQL

各テーブルでRLSを有効にし、SELECT, UPDATE, DELETE, INSERT で利用されるポリシーを定義していきます。

ALTER TABLE table_name ENABLE ROW LEVEL SECURITY;

CREATE POLICY policy_name ON table_name
    FOR SELECT USING (
         CASE 
              WHEN current_setting('app.current_team', true) IS NULL OR current_setting('app.current_team', true) = '' THEN 
                   false
              ELSE 
                   team_id = current_setting('app.current_team', true)::uuid
         END
    );

CASE文でちょっとややこしくしていますが、要点としては current_setting で取得できる app.current_name という実行時設定パラメータが、対象テーブルの team_id と一致するときのみアクセスを許可する、という内容です。

CASE文を利用している理由ですが、

• team_id は UUID型で定義しているため、app.current_teamで取得されたtextをUUIDにキャストする必要があった。
• 実行時設定パラメータが設定されない場合、current_setting('app.current_team', true) はNULL（空文字になることも？） になるため、このとき UUIDにキャストしようとすると実行エラーになる。
• 設定がない場合には常にアクセスを許可しない。

ということで上記のようになっています。

SQLAlchemyでの運用

SQLAlchemy (SQLModel) を利用して、各Sessionごとに team_id を設定します。
例えばアクセストークンを解析して取得したuser_idから、ユーザーが所属しているteam_idを取得し、それをセッションごとに設定する、という運用になります。

DBへの接続は、postgresロールを使わずに先ほど作成した新しいユーザーを使ってください。

def set_team_id(team_id: UUID, session: Session):
    """
    RLSで使用するチームIDをセットする
    """
    session.execute(text(f"SET LOCAL app.current_team = '{team_id}'"))
    return

テストコードを書いて、RLSが設定されていることをテストします。

from sqlmodel import Session, delete, select
from lib.infra.db import set_team_id
from uuid import uuid4

from lib.models import *


def test_rls_with_no_team_id(test_engine):
    with Session(test_engine) as session:
        session.exec(delete(MTeam))

        new_team = MTeam(team_name="test_team")
        session.add(new_team)
        session.commit()
        session.refresh(new_team)
        team_id = new_team.team_id

    with Session(test_engine) as session:
        set_team_id(team_id=team_id, session=session)
        # projectを作成
        new_project = TProject(
            project_name="test_project",
            team_id=team_id,
        )
        session.add(new_project)
        session.flush()

        session.commit()

    with Session(test_engine) as session:
        set_team_id(team_id=uuid4(), session=session)
        projects = session.exec(select(TProject)).all()

        assert len(projects) == 0


def test_rls_with_team_id(test_engine):
    with Session(test_engine) as session:
        session.exec(delete(MTeam))

        new_team = MTeam(team_name="test_team")
        session.add(new_team)
        session.commit()
        session.refresh(new_team)
        team_id = new_team.team_id

    with Session(test_engine) as session:
        set_team_id(team_id=team_id, session=session)
        # projectを作成
        new_project = TProject(
            project_name="test_project",
            team_id=team_id,
        )
        session.add(new_project)
        session.flush()

        session.commit()

    with Session(test_engine) as session:
        set_team_id(team_id=team_id, session=session)
        projects = session.exec(select(TProject)).all()

        assert len(projects) == 1

    with Session(test_engine) as session:
        set_team_id(team_id=uuid4(), session=session)
        projects = session.exec(select(TProject)).all()

        assert len(projects) == 0

大量のテーブルに対して同じRLSを設定する

supabaseのGUIからポチポチしてポリシー設定したり、テーブルごとにsqlを書いて設定するのもやりたくありません。基本となるRLS設定は変わらないため、同じSQLで対象となる全てのテーブルに設定することを考えます。

以下のように テーブル名の配列を記述し、LOOP処理で同じSQLを実行していきます。

DO $$ 
DECLARE 
      -- テーブル名の配列
     table_names TEXT[] := ARRAY[
          't_projects',
          't_devices'
     ]; 
     table_name TEXT;
     select_policy_name TEXT;
     update_policy_name TEXT;
     delete_policy_name TEXT;
     insert_policy_name TEXT;
BEGIN
     -- テーブル名のリストを LOOP で処理
     FOREACH table_name IN ARRAY table_names
     LOOP
          RAISE NOTICE 'Processing table: %', table_name;

          select_policy_name := format('%I_select_policy', table_name);
          update_policy_name := format('%I_update_policy', table_name);
          delete_policy_name := format('%I_delete_policy', table_name);
          insert_policy_name := format('%I_insert_policy', table_name);

          EXECUTE format(
               'ALTER TABLE %I ENABLE ROW LEVEL SECURITY;',
               table_name
          );

          EXECUTE format(
               'DROP POLICY IF EXISTS %I ON %I;',
               select_policy_name,
               table_name
          );

          EXECUTE format(
               'CREATE POLICY %I ON %I
                    FOR SELECT USING (
                         CASE 
                              WHEN current_setting(''app.current_team'', true) IS NULL OR current_setting(''app.current_team'', true) = '''' THEN 
                                   false
                              ELSE 
                                   team_id = current_setting(''app.current_team'', true)::uuid
                         END
                    );',
               select_policy_name,
               table_name
          );

          EXECUTE format(
               'DROP POLICY IF EXISTS %I ON %I;',
               update_policy_name,
               table_name
          );

          EXECUTE format(
               'CREATE POLICY %I ON %I
                    FOR UPDATE USING (
                         CASE 
                              WHEN current_setting(''app.current_team'', true) IS NULL OR current_setting(''app.current_team'', true) = '''' THEN 
                                   false
                              ELSE 
                                   team_id = current_setting(''app.current_team'', true)::uuid
                         END
                    );',
               update_policy_name,
               table_name
          );

          
          EXECUTE format(
               'DROP POLICY IF EXISTS %I ON %I;',
               delete_policy_name,
               table_name
          );

          EXECUTE format(
               'CREATE POLICY %I ON %I
                    FOR DELETE USING (
                         CASE 
                              WHEN current_setting(''app.current_team'', true) IS NULL OR current_setting(''app.current_team'', true) = '''' THEN 
                                   false
                              ELSE 
                                   team_id = current_setting(''app.current_team'', true)::uuid
                         END
                    );',
               delete_policy_name,
               table_name
          );

          EXECUTE format(
               'DROP POLICY IF EXISTS %I ON %I;',
               insert_policy_name,
               table_name
          );

          EXECUTE format(
               'CREATE POLICY %I ON %I
                    FOR INSERT WITH CHECK (
                         CASE 
                              WHEN current_setting(''app.current_team'', true) IS NULL OR current_setting(''app.current_team'', true) = '''' THEN 
                                   false
                              ELSE 
                                   team_id = current_setting(''app.current_team'', true)::uuid
                         END
                    );',
               insert_policy_name,
               table_name
          );

     END LOOP;
END $$;