Row Level Security導入のためにやったこと

はじめに

業務委託としてお手伝いをしている渡部です。

Stena Expenseは、1つのソフトウェアを様々なお客様がご利用される、いわゆるマルチテナント型のSaaSです。
マルチテナント型のSaaSでは、テナントごとにアプリケーションやデータベースをどのように分離するか、もしくは分離しないかが焦点になりますが、先日Stena Expenseではデータベースの分離方法の1つであるRow Level Security(以下、RLS)を導入したため、導入背景や作業時に実施したことを紹介します。

アーキテクチャの分離レベルとRLS導入の背景

マルチテナントのアーキテクチャを検討するにあたり、複数の分離レベルが存在します。

1. アプリケーションとデータベースをテナント間で共有し、テナントごとに分離しない
2. アプリケーションとデータベースのテーブルをテナント間で共有し、テナントごとにテーブルの行レベルで分離する(RLS)
3. アプリケーションとデータベースをテナント間で共有し、テナントごとにデータベースのスキーマレベルで分離する
4. アプリケーションをテナント間で共有し、テナントごとにデータベースの実行環境レベルで分離する
5. テナントごとにアプリケーションの実行環境とデータベースの実行環境どちらも分離する

数字が大きくなるにれて、分離レベルは強くなりますが、その一方で移行コストや運用コストは高くなることが想定されます。

Stena Expenseではこれまで、1の分離レベルを採用していました。そのため、データベースへのクエリ実行時に、以下のようなクエリを実行しています。(以下の説明では、テナントを区別するカラムとして、tenant_idを利用します。)

SELECT * FROM example_resources WHERE tenant_id = 'test'

ここで問題となるのが、WHERE句に指定しているWHERE tenant_id = 'test'です。もし、このWHERE句の指定を忘れてしまうと、あるテナントAから別のテナントBのリソースが閲覧できてしまうなどの重大なセキュリティインシデントにつながる可能性があり、リリース前に社内で実施している脆弱性診断で修正事項として報告されます。一方で、WHERE句の指定漏れをPRのレビューなどによって人力で防ぐには限りがあります。このようなインシデントを仕組みで防ぐため、Stena Expenseではデータベースの分離方法を検討するに至りました。

2~5番を含むさまざまな方法を検討した結果、2のRLSを導入することに決定しました。
RLSを採用した理由は、すでに運用中のアプリケーションのデータベースを分離する場合に3以上の方法を採用すると多大な移行コストや運用コストが見込まれたためです。

RLS導入のために実施したこと

RLSの対象テーブルの洗い出し

まず、RLSの導入にあたり、RLS適用対象テーブルを洗い出しました。基本的には、tenant_idカラムを持つテーブルをRLSの対象としました。

RLS対象テーブルの子テーブルへのtenant_idカラムの追加

既存のテーブル設計においては、親テーブルのみがtenant_idカラムを持っていました。しかし、今後子テーブルからもテナントごとにレコードを抽出する可能性がある点やアプリケーションの処理単位もテナント単位であることを考慮し、子テーブルへもtenant_idカラムを追加し、さらにはRLSの対象とすることにしました。

RLS対象のAPI、Batchの洗い出し

APIやBatchによっては、複数のテナントのデータを対象に操作をしたい場合が存在しました。
例えば、社内のカスタマーサポートのメンバーが利用する管理画面には、複数のテナントへ向けたお知らせの配信機能などが存在します。こうした機能を実現するためのAPIでは、複数のテナント向けのデータを処理しているため、RLSの対象とすることは出来ません。
また、Batchにおいても複数のテナントを対象に集計する処理などが存在するため、RLSの対象とすることはできないものも存在します。
このように、API、BatchレベルでもRLSの対象可否について洗い出しを行いました。

テナント分離の観点からは、Batchでもテナント毎に処理をすることも考えられますが、対応工数やパフォーマンスの観点から、この方法は保留としました。

データベース用ROLEの作成

RLSを利用するために、RLSをバイパスしないオプションであるNOBYPASSRLSオプションが付与されたROLEを作成しました。
ROLE作成時は、複数のALTERステートメントを実行し、TABLESやSEQUENCESへの権限を付与したのですが、今思うとPostgreSQL 14から利用できるようになったPredefined ROLESである、pg_read_all_dataとpg_write_all_dataを利用すれば、より簡単にROLEの権限付与が行えたかもしれません。

作成したROLEは、先のRLS対象の洗い出しに基づいて、以下のように使い分けを行っています。

• RLS対象の処理では、NOBYPASSRLSオプションが付与されたRLS用のROLEを使用する
• RLS対象外の処理では、BYPASSRLSオプションが付与されたROLEを使用する

補足ですが、ローカルのDBとクラウド上のDBで、ROLE名は同じでも付与されているオプションが異なる場合が存在しました。既存のROLEを使い回す場合は、各環境で以下のようなクエリを実行し、ROLEに付与されたオプションを確認するとよいでしょう。

SELECT rolname, rolbypassrls FROM pg_roles WHERE rolname = 'test_user';

NOBYPASSRLS, BYPASSRLSオプションについては、公式ドキュメントが参考になります。
https://www.postgresql.jp/document/16/html/sql-createrole.html

RLSの設定

RLSの適用対象テーブルごとに、RLSを適用します。以下はRLS適用時のステートメントの例です。

ALTER TABLE example_resources ENABLE ROW LEVEL SECURITY;
CREATE POLICY rls_policy_example_resources ON example_resources USING (tenant_id = current_setting('app.tenant_id'));

RLSを利用する場合、PostgreSQLのcurrent_userと使う方法とcurrent_settingを使う方法が考えられますが、current_userを使う場合は以下のように運用面で課題があります。

• テナントごとにPostgreSQLのロールを用意する必要がある
• コネクション数の問題が発生し得る(テナントごとにロールを使いデータベースへ接続するため)

一方で、current_settingはセッションごとに値を設定・参照でき上記の課題を解決しているため、current_settingの使用を採用しました。

current_userやcurrent_settingを使う方法については、以下記事参考になりました。
https://aws.amazon.com/jp/blogs/news/multi-tenant-data-isolation-with-postgresql-row-level-security/

RLSの利用(GORMの例)

ORMにGORMを利用しているため、GORMを例にサンプルコードを用いてRLSの利用方法について紹介します。

contextへのtenant_idの設定

contextへのtenant_idの設定は認証用のミドルウェア等で行っています。

type TenantIDKey struct{}

func (a exampleMiddleware) auth(ctx context.Context) (context.Context, error) {
	// some implementations

	user, err := exampleRepository.Get(ctx, userID)
	if err != nil {
		return ctx, nil
	}
	tenantID := user.Tenant.ID

	newCtx = context.WithValue(ctx, TenantIDKey{}, tenantID)

	return newCtx, nil
}

current_settingへの値の設定

current_settingへの値の設定は、「コネクションプールを作成するとき」と「コネクションプールを再利用するとき」に自動的に設定されるようにしています。
具体的には、database/sqlのSessionResetterとConnectorのインターフェースを実装することで実現しています。

package example

import (
	"context"
	"database/sql/driver"
)

type connector struct {
	dsn    string
	driver driver.Driver
}

type conn struct {
	driver.Conn
}

// Connect database/sqlにて、新しいconnectionが生成されるときに呼ばれる
func (c *connector) Connect(ctx context.Context) (driver.Conn, error) {
	cn, err := c.driver.Open(c.dsn)
	if err != nil {
		return nil, err
	}

	err = setTenantID(ctx, cn)
	if err != nil {
		return nil, err
	}

	return &conn{cn}, nil
}

func (c *connector) Driver() driver.Driver {
	return c.driver
}

// ResetSession database/sqlにて、ConnectionPoolからconnectionが取り出されるときに呼ばれる
func (c *conn) ResetSession(ctx context.Context) error {
	err := setTenantID(ctx, c)
	if err != nil {
		return err
	}

	return nil
}

// setTenantID current_settingのapp.tenant_idにcontextから取り出したtenant_idをセットする
func setTenantID(ctx context.Context, cn driver.Conn) error {
	tenantID := ctx.Value(TenantIDKey{})
	if tenantID == "" {
		return nil
	}

	stmt, err := cn.Prepare(fmt.Sprintf("SET app.tenant_id = '%s'", tenantID.ToString()))
	if err != nil {
		return err
	}

	args := []driver.NamedValue{}
	_, err = stmt.(driver.StmtExecContext).ExecContext(ctx, args)
	if err != nil {
		return err
	}

	return nil
}

自作したconnectorはGORMの初期化時に以下のように利用しています。database/sqlのOpenDB関数を使用することで、*sql.DBを生成し、gormの初期化時に渡します。

func connectDB() (*gorm.DB, error) {
  // some implementations

	connector := &connector{dsn: address, driver: &stdlib.Driver{}}
	sqlDB := sql.OpenDB(connector)

	db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), gormConfig)

  // some implementations
}

これらの実装は、以下の記事が参考になりました。
https://times.hrbrain.co.jp/entry/postgresql-row-level-security-go

クエリの実行

GORMを利用したクエリの実行について、クエリの改修前後のイメージはこちらです。

改修前のコード

func GetExampleResourceByTenantID(tenantID string) (ExampleResource, error) {
	exampleResource := ExampleResource{}
	err := *gorm.DB.Where("tenant_id = ?", tenantID.ToString()).Find(&exampleResource).Error
	return exampleResource, err
}

改修前は、引数にtenantIDが渡され、Where関数で絞り込みを行っています。

改修後のコード

func GetExampleResourceByTenantID(ctx context.Context) (ExampleResource, error) {
	exampleResource := ExampleResource{}
	err := *gorm.DB.WithContext(ctx).Find(&exampleResource).Error
	return exampleResource, err
}

改修後は、WitContext関数を利用し、引数のcontextを渡しています。これにより、クエリの実行前に自前で実装したConnectやResetSessionが実行され、contextに設定されたtenant_idを参照し、current_settingにtenant_idが設定されます。結果として、Where関数によるtenant_idの絞り込みを行わなくても、RLSによりtenant_idで分離した状態でクエリをすることができるようになっています。

このWithContextは、すべてのGORMのメソッドの呼び出し時は必ず実行するようにしています。本来的には、テーブルごとのRLSの利用可否によりWithContextの利用可否を決められますが、認知コストが高くなるため、このような運用をしています。
また、改修後のコードではわかりやすさを重視し、tenant_idの絞り込みを行うWhere関数を削除していますが、実際は削除せず残しています。これは、意図せずRLSが効かなかった場合もテナントのデータを保護することを目的としています。

CIによるRLSの設定漏れの検知

RLSを有効化し運用していくためには、様々な設定が必要になることがおわかりいただけたかと思います。これらの設定漏れが発生してしまうとインシデントに繋がりうるため、CIで自動的に検知できるようにしました。

• GORMのメソッド呼び出し時に、WithContextメソッドが呼び出されていない場合はFailにする
  • Goのanalysisパッケージを使い静的解析を実装しています
• データベースのテーブルに変更があった場合に、「RLSの設定漏れがある場合」もしくは「tenant_idの追加漏れがある場合」はFailにする
  • データベースのスキーマをPythonで解析し、設定漏れを検知しています

本番環境導入前の検証

RLSを本番環境に導入する前に、複数の観点から開発環境での検証を行いました。
※ここでの開発環境とは、本番環境と同等のインフラ構成のクラウド上の環境を指しています。

Feature FlagによるRLSの有効化

まず、本番環境には影響を与えずに開発環境でRLSの検証を行うために、Feature Flagを利用してRLSの使用可否を切り替えられるようにしました。アプリケーション側のDB接続時のコードにて環境変数を用いてFeature Flagを実装しています。

パフォーマンステスト

RLSは内部的にtenant_idによる絞り込みが行われるため、パフォーマンスが悪化する場合もあるため、負荷の高いクエリについては個別にパフォーマンステストを行いました。
まず、データベースにはAlloyDBを利用しているため、Query Insightから負荷が高いクエリを特定しました。次に、開発環境にてRLSの適用前後で負荷の高いクエリを実行し、パフォーマンス影響の有無の確認を行いました。
結果としては、RLS適用によるパフォーマンス影響がほぼなかったため、新しいインデックスを追加するなどの作業は発生しませんでした。

ホワイトボックステスト/ブラックボックステスト

重要機能については、テストケースを元に手動テストを実施し、RLSによるエラーが発生しないことを確認しました。
また、開発環境は他の用途(本番環境リリース前の機能のリリース判定など)にも頻繁に利用されるため、1週間程度RLSを適用した状態でログをモニタリングし、問題が発生しないことを確認しました。

こうした検証により、本番環境でのトラブルを起こさずRLSの導入が出来ました。

最後に

Row Level Security導入のために実施したことは以上となります。
今後のデータ量増加によって、インデックスの追加やパーティショニングなどさまざまなアプローチを探っていきたいと思います。

参考

• https://docs.aws.amazon.com/ja_jp/whitepapers/latest/saas-architecture-fundamentals/saas-architecture-fundamentals.html
• https://speakerdeck.com/yudppp/row-level-security-is-silver-bullet-for-multitenancy
• https://times.hrbrain.co.jp/entry/postgresql-row-level-security
• https://www.youtube.com/live/L9Ek9g_oU9c?si=RnkU6teIKV5pLiqx