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になっています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/494791/b54be126-c252-9f15-3ae6-85205b984d2d.png)

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

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

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


# RLSを設定するSQL

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

```sql
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ロールを使わずに先ほど作成した新しいユーザーを使ってください。

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

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


```py
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を実行していきます。

```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 $$;
```