Postgres 行レベルセキュリティ (RLS) の限界と代替

-- RLS ポリシー
CREATE POLICY tenant_isolation ON orders
FOR ALL TO app_user
USING (tenant_id = current_setting('app.tenant_id')::int);

-- このクエリは:
-- 1. orders テーブルをスキャン
-- 2. 各行について tenant_id が現在の設定と一致するか確認
-- 3. ユーザーの WHERE 句を適用
-- 4. 一致した行を返す
SELECT * FROM orders WHERE status = 'pending';

-- 複数ポリシーは評価コストを積み上げる
CREATE POLICY tenant_policy ON orders
USING (tenant_id = current_setting('app.tenant_id')::int);

CREATE POLICY department_policy ON orders
USING (department_id IN (
  SELECT dept_id FROM user_departments
  WHERE user_id = current_setting('app.user_id')::int
));

-- すべての行が両ポリシーチェックを通過しなければならない

-- セキュリティビュー
CREATE VIEW tenant_orders AS
SELECT * FROM orders
WHERE tenant_id = current_setting('app.tenant_id')::int;

-- ビューに対するクエリ
SELECT * FROM tenant_orders WHERE status = 'pending';

-- PostgreSQL は次のように最適化できる:
-- SELECT * FROM orders
-- WHERE tenant_id = current_setting('app.tenant_id')::int
--   AND status = 'pending'
-- そして (tenant_id, status) の複合索引を使う

-- RLS の場合: 行取得後にセキュリティチェックが走るため、
-- tenant_id の索引が最適に使われないことがある

-- ビューの場合: (tenant_id, status) の複合索引を有効に使える
CREATE INDEX idx_orders_tenant_status ON orders(tenant_id, status);

-- RLS でできるのはこの程度:
CREATE POLICY simple_tenant_policy ON documents
USING (tenant_id = current_setting('app.tenant_id')::int);

-- これは無理:
-- 「ユーザーは、同じ部署で
--   かつ業務時間内で
--   かつ必要な研修を完了していて
--   かつドキュメントがコンテンツモデレーションでフラグされておらず
--   かつ外部コンプライアンス API がアクセスを承認した場合に、
--   ドキュメントへアクセスできる」

-- セッションコンテキストに限定
CREATE POLICY time_based_policy ON sensitive_data
USING (
  EXTRACT(hour FROM NOW()) BETWEEN 9 AND 17 -- 業務時間の粗い近似
  AND EXTRACT(dow FROM NOW()) BETWEEN 1 AND 5 -- 平日のみ
);

-- 考慮できない:
-- - ユーザーの実タイムゾーンや所在
-- - 企業固有の休日カレンダー
-- - ユーザーのデバイス/IP レピュテーション
-- - 直近のセキュリティイベントやリスクスコア
-- - 動的なコンプライアンス要件

-- これは RLS で不可能:
-- CREATE POLICY compliance_policy ON financial_records
-- USING (external_compliance_service_check(user_id, document_id, current_timestamp));

-- 制約は PostgreSQL インスタンス内のデータに限られる

// Casbin など ABAC エンジンを利用
const authorizationEngine = {
  async checkAccess(subject, resource, action, context) {
    const policy = await this.evaluatePolicy({
      // ユーザー属性
      user: {
        id: subject.id,
        roles: subject.roles,
        department: subject.department,
        clearanceLevel: subject.clearanceLevel,
        lastTrainingDate: subject.lastTrainingDate,
        currentLocation: await geoService.getLocation(context.ip),
        riskScore: await riskEngine.getUserRiskScore(subject.id),
      },

      // リソース属性
      resource: {
        id: resource.id,
        classification: resource.dataClassification,
        owner: resource.owner,
        createdDate: resource.createdDate,
        lastModified: resource.lastModified,
        complianceFlags: await complianceService.getFlags(resource.id),
      },

      // 環境属性
      environment: {
        currentTime: new Date(),
        userAgent: context.userAgent,
        networkLocation: context.networkLocation,
        isBusinessHours: await businessCalendar.isBusinessHours(),
        threatLevel: await securityService.getCurrentThreatLevel(),
      },

      action: action,
    });

    return policy.decision;
  },
};

// 利用
const canAccess = await authorizationEngine.checkAccess(
  currentUser,
  requestedDocument,
  'read',
  requestContext,
);