PostgreSQLの高度なロールベースアクセス制御: 行レベルセキュリティとポリシー管理

はじめに

PostgreSQLの行レベルセキュリティ（RLS）は、行レベルでのきめ細かなアクセス制御を提供し、データベース内で複雑な認証ルールを強制することを可能にします。これにより、アプリケーション層でのフィルタリングの必要性がなくなり、アプリケーションがデータベースにどのようにアクセスしてもデータセキュリティを確保できます。

なぜ行レベルセキュリティが必要なのか？

従来のアプローチの問題点

アプリケーション層でのフィルタリング:

# すべてのクエリでフィルタリングを忘れずに記述する必要がある
query = "SELECT * FROM documents WHERE tenant_id = %s"
results = db.execute(query, [current_user_tenant_id])

問題点:
- フィルタリングを忘れやすい（セキュリティ脆弱性）
- アプリケーション全体でコードが重複する
- 直接的なデータベースアクセスに対して保護されない
- 複雑な監査要件

行レベルセキュリティの利点

• 自動的な強制: ポリシーが透過的に適用される
• 一元化されたルール: セキュリティロジックが一箇所に集約される
• マルチテナント分離: SaaSアプリケーションに最適
• 監査コンプライアンス: 組み込みのセキュリティ保証
• パフォーマンス: 効率的なクエリ計画

基本的なRLS設定

ステップ1: テーブルでRLSを有効化

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(200),
    content TEXT,
    owner_id INTEGER NOT NULL,
    department VARCHAR(50),
    is_public BOOLEAN DEFAULT false,
    created_at TIMESTAMP DEFAULT NOW()
);

-- 行レベルセキュリティを有効化
ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

ステップ2: ロールの作成

-- 異なるユーザロールを作成
CREATE ROLE regular_user;
CREATE ROLE department_manager;
CREATE ROLE admin;

-- テーブルへのアクセスを許可
GRANT SELECT, INSERT, UPDATE, DELETE ON documents TO regular_user;
GRANT SELECT, INSERT, UPDATE, DELETE ON documents TO department_manager;
GRANT ALL ON documents TO admin;

ステップ3: ポリシーの作成

-- ポリシー: ユーザは自分自身のドキュメントを見られる
CREATE POLICY user_own_documents ON documents
    FOR SELECT
    TO regular_user
    USING (owner_id = current_setting('app.user_id')::INTEGER);

-- ポリシー: ユーザは自分自身が所有するドキュメントを挿入できる
CREATE POLICY user_insert_own ON documents
    FOR INSERT
    TO regular_user
    WITH CHECK (owner_id = current_setting('app.user_id')::INTEGER);

-- ポリシー: ユーザは自分自身のドキュメントを更新できる
CREATE POLICY user_update_own ON documents
    FOR UPDATE
    TO regular_user
    USING (owner_id = current_setting('app.user_id')::INTEGER)
    WITH CHECK (owner_id = current_setting('app.user_id')::INTEGER);

ステップ4: セッション変数の設定

アプリケーションコード内:

import psycopg2

# ロールとして接続
conn = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="regular_user",
    password="password"
)

# ユーザコンテキストを設定
cursor = conn.cursor()
cursor.execute("SET app.user_id = %s", [current_user_id])

# これ以降のクエリは自動的にフィルタリングされる
cursor.execute("SELECT * FROM documents")
# owner_id = current_user_id であるドキュメントのみが返される

マルチテナントSaaSアプリケーション

完全なマルチテナント設定

-- テナントテーブルを作成
CREATE TABLE tenants (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

-- ユーザテーブルを作成
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) NOT NULL,
    tenant_id INTEGER REFERENCES tenants(id),
    role VARCHAR(20) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

-- アプリケーションデータテーブルを作成
CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    tenant_id INTEGER NOT NULL REFERENCES tenants(id),
    customer_name VARCHAR(100),
    amount DECIMAL(10,2),
    status VARCHAR(20),
    created_by INTEGER REFERENCES users(id),
    created_at TIMESTAMP DEFAULT NOW()
);

-- RLSを有効化
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;

-- ポリシー: ユーザは自テナントの注文のみを見られる
CREATE POLICY tenant_isolation ON orders
    FOR ALL
    TO public
    USING (tenant_id = current_setting('app.tenant_id')::INTEGER)
    WITH CHECK (tenant_id = current_setting('app.tenant_id')::INTEGER);

アプリケーション統合

Djangoの例:

from django.db import connection

class TenantMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        if request.user.is_authenticated:
            with connection.cursor() as cursor:
                cursor.execute(
                    "SET app.tenant_id = %s",
                    [request.user.tenant_id]
                )

        response = self.get_response(request)
        return response

Node.jsの例:

const { Pool } = require('pg');
const pool = new Pool();

async function setTenantContext(tenantId) {
    const client = await pool.connect();
    try {
        await client.query('SET app.tenant_id = $1', [tenantId]);
        return client;
    } catch (err) {
        client.release();
        throw err;
    }
}

// 使用法
app.use(async (req, res, next) => {
    if (req.user) {
        req.dbClient = await setTenantContext(req.user.tenantId);
    }
    next();
});

高度なポリシーパターン

1. 階層型アクセス制御

-- ユーザは所属部署のドキュメントと公開ドキュメントを見られる
CREATE POLICY department_access ON documents
    FOR SELECT
    TO regular_user
    USING (
        department = current_setting('app.user_department', true)
        OR is_public = true
        OR owner_id = current_setting('app.user_id')::INTEGER
    );

2. 時間ベースのアクセス

CREATE TABLE subscriptions (
    id SERIAL PRIMARY KEY,
    user_id INTEGER NOT NULL,
    content TEXT,
    valid_from TIMESTAMP NOT NULL,
    valid_until TIMESTAMP NOT NULL
);

ALTER TABLE subscriptions ENABLE ROW LEVEL SECURITY;

-- ポリシー: ユーザはアクティブなサブスクリプションのみを見られる
CREATE POLICY active_subscriptions ON subscriptions
    FOR SELECT
    TO public
    USING (
        user_id = current_setting('app.user_id')::INTEGER
        AND NOW() BETWEEN valid_from AND valid_until
    );

3. 複雑なロジックを含むロールベースアクセス

CREATE TABLE projects (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100),
    owner_id INTEGER,
    team_id INTEGER,
    visibility VARCHAR(20) -- 'private', 'team', 'public'
);

ALTER TABLE projects ENABLE ROW LEVEL SECURITY;

-- 複雑な可視性ポリシー
CREATE POLICY project_visibility ON projects
    FOR SELECT
    TO public
    USING (
        CASE current_setting('app.user_role', true)
            WHEN 'admin' THEN true
            WHEN 'manager' THEN (
                team_id = current_setting('app.user_team_id')::INTEGER
                OR visibility = 'public'
            )
            ELSE (
                owner_id = current_setting('app.user_id')::INTEGER
                OR visibility = 'public'
            )
        END
    );

4. 共有アクセスリスト

CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title VARCHAR(200),
    owner_id INTEGER NOT NULL
);

CREATE TABLE document_shares (
    document_id INTEGER REFERENCES documents(id),
    shared_with_user_id INTEGER,
    permission VARCHAR(20), -- 'read', 'write'
    PRIMARY KEY (document_id, shared_with_user_id)
);

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

-- ポリシー: ユーザは所有または共有されたドキュメントを見られる
CREATE POLICY document_access ON documents
    FOR SELECT
    TO public
    USING (
        owner_id = current_setting('app.user_id')::INTEGER
        OR id IN (
            SELECT document_id 
            FROM document_shares 
            WHERE shared_with_user_id = current_setting('app.user_id')::INTEGER
        )
    );

-- ポリシー: ユーザは所有するドキュメント、または書き込み権限で共有されたドキュメントを変更できる
CREATE POLICY document_modify ON documents
    FOR UPDATE
    TO public
    USING (
        owner_id = current_setting('app.user_id')::INTEGER
        OR id IN (
            SELECT document_id 
            FROM document_shares 
            WHERE shared_with_user_id = current_setting('app.user_id')::INTEGER
              AND permission = 'write'
        )
    );

管理タスクのためのRLSのバイパス

オプション1: BYPASSRLSロール属性

-- スーパーーユーザーのようなアクセスを許可
CREATE ROLE admin_user WITH LOGIN PASSWORD 'secure_password';
ALTER ROLE admin_user BYPASSRLS;

-- 管理者はポリシーに関係なくすべての行を見られる

オプション2: 管理者向けの許容ポリシー

-- 管理者ロール用の許容ポリシーを作成
CREATE POLICY admin_all_access ON documents
    FOR ALL
    TO admin
    USING (true)
    WITH CHECK (true);

オプション3: 一時的にRLSを無効化

-- メンテナンススクリプト用
BEGIN;
    ALTER TABLE documents DISABLE ROW LEVEL SECURITY;
    -- 管理操作を実行
    UPDATE documents SET archived = true WHERE created_at < NOW() - INTERVAL '1 year';
    ALTER TABLE documents ENABLE ROW LEVEL SECURITY;
COMMIT;

ポリシーの種類: USING 対 WITH CHECK

USING 句

既存の行のうち、どの行が表示/変更可能かを制御します:

CREATE POLICY select_own ON documents
    FOR SELECT
    USING (owner_id = current_user_id());

-- ユーザはowner_idが一致する行のみをSELECTできる

WITH CHECK 句

新規/更新された行のうち、どの行が許可されるかを制御します:

CREATE POLICY insert_own ON documents
    FOR INSERT
    WITH CHECK (owner_id = current_user_id());

-- ユーザはowner_idが一致する行のみをINSERTできる

結合されたポリシー

CREATE POLICY update_own ON documents
    FOR UPDATE
    USING (owner_id = current_user_id())       -- 自分自身の行のみ更新可能
    WITH CHECK (owner_id = current_user_id()); -- 所有権を変更できない

パフォーマンスに関する考慮事項

1. ポリシーで使用されるカラムにインデックスを作成

-- ポリシーで使用されるカラムにインデックスを作成
CREATE INDEX idx_documents_owner ON documents(owner_id);
CREATE INDEX idx_documents_tenant ON documents(tenant_id);
CREATE INDEX idx_documents_department ON documents(department);

2. クエリ計画の分析

EXPLAIN ANALYZE
SELECT * FROM documents
WHERE title LIKE '%report%';

-- ポリシーフィルタが効率的に適用されているか確認

3. ポリシー内でのサブクエリの回避

悪い例 - 各行に対してサブクエリが実行される:

CREATE POLICY slow_policy ON documents
    USING (
        id IN (SELECT document_id FROM shares WHERE user_id = current_user_id())
    );

より良い例 - EXISTSを使用:

CREATE POLICY faster_policy ON documents
    USING (
        EXISTS (
            SELECT 1 FROM shares 
            WHERE document_id = documents.id 
              AND user_id = current_user_id()
        )
    );

最適な例 - アプリケーションでJOINを使用、またはマテリアライズドビューを使用:

CREATE MATERIALIZED VIEW user_accessible_documents AS
SELECT d.*, s.user_id AS shared_with
FROM documents d
LEFT JOIN shares s ON d.id = s.document_id;

CREATE INDEX ON user_accessible_documents(shared_with);

監視と監査

アクティブなポリシーの表示

-- すべてのRLSポリシーを一覧表示
SELECT 
    schemaname,
    tablename,
    policyname,
    permissive,
    roles,
    cmd,
    qual,
    with_check
FROM pg_policies
ORDER BY tablename, policyname;

ポリシー変更の監査

-- ポリシー変更用の監査ログを作成
CREATE TABLE policy_audit_log (
    id SERIAL PRIMARY KEY,
    table_name VARCHAR(100),
    policy_name VARCHAR(100),
    action VARCHAR(20),
    changed_by VARCHAR(50),
    changed_at TIMESTAMP DEFAULT NOW()
);

-- 変更を記録するトリガー関数
CREATE OR REPLACE FUNCTION log_policy_changes()
RETURNS event_trigger AS $$
BEGIN
    INSERT INTO policy_audit_log (table_name, policy_name, action, changed_by)
    SELECT 
        objid::regclass::text,
        object_identity,
        'CREATED',
        current_user
    FROM pg_event_trigger_ddl_commands()
    WHERE command_tag = 'CREATE POLICY';
END;
$$ LANGUAGE plpgsql;

CREATE EVENT TRIGGER log_policy_create
    ON ddl_command_end
    WHEN TAG IN ('CREATE POLICY')
    EXECUTE FUNCTION log_policy_changes();

ポリシーの効果テスト

-- テスト関数を作成
CREATE OR REPLACE FUNCTION test_user_access(
    test_user_id INTEGER,
    test_tenant_id INTEGER
)
RETURNS TABLE(document_id INTEGER, title VARCHAR) AS $$
BEGIN
    -- セッション変数を設定
    PERFORM set_config('app.user_id', test_user_id::text, true);
    PERFORM set_config('app.tenant_id', test_tenant_id::text, true);

    -- アクセス可能なドキュメントを返却
    RETURN QUERY SELECT id, documents.title FROM documents;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

-- アクセステスト
SELECT * FROM test_user_access(123, 5);

トラブルシューティング

問題1: ポリシーが適用されない

RLSが有効か確認:

SELECT tablename, rowsecurity 
FROM pg_tables 
WHERE schemaname = 'public' AND tablename = 'documents';

無効な場合に有効化:

ALTER TABLE documents ENABLE ROW LEVEL SECURITY;

問題2: セッション変数が設定されていない

現在の設定をデバッグ:

SHOW app.user_id;
SHOW app.tenant_id;

-- または current_setting を使用
SELECT current_setting('app.user_id', true);

問題3: パフォーマンスの低下

スローなポリシーを特定:

EXPLAIN (ANALYZE, BUFFERS) 
SELECT * FROM documents WHERE title = 'test';

-- ポリシー条件を含む「Filter」操作を探す

解決策: ポリシーを簡略化、またはインデックスを追加

問題4: 予期しないアクセス拒否

特定のユーザとしてテスト:

SET ROLE regular_user;
SET app.user_id = '123';

SELECT * FROM documents;
-- そのユーザが実際に見られる内容を表示

RESET ROLE;

ベストプラクティス

1. 常にセッション変数を設定: current_setting() の値が入っていることを確認する
2. ポリシーで使用されるカラムにインデックスを作成: USING/WITH CHECK で使用されるすべてのカラムにインデックスを作成する
3. 徹底的にテスト: 異なるロールとシナリオでポリシーを検証する
4. ポリシーをシンプルに保つ: 複雑なロジックはパフォーマンスに影響を与える
5. EXPLAINを使用: RLSを有効にしてクエリ計画を分析する
6. ポリシーをドキュメント化: メンテナンス性のためにポリシーロジックにコメントを付ける
7. 関心事を分離: SELECT、INSERT、UPDATE、DELETEのために異なるポリシーを使用する
8. パフォーマンスを監視: スローなクエリのアラートを設定する

まとめ

PostgreSQLの行レベルセキュリティは、堅牢でデータベース内で強制されるアクセス制御を提供します:

• 自動的なフィルタリング: セキュリティが透過的に適用される
• マルテナント分離: SaaSアプリケーションに最適
• 柔軟なポリシー: 複雑な認証ルールをサポート
• 一元化されたセキュリティ: ロジックはデータベース内にあり、コード内に散らばっていない

まずはシンプルなテナント分離から始め、パフォーマンスへの影響をテストし、必要に応じてより高度な認証パターンへと拡張していきましょう。