AWS Cognito + Terraform で実現する複雑な認証・認可設計：8 種類のユーザー属性を管理するエンタープライズ SaaS の実践

はじめに：B2B SaaS の認証・認可はなぜ複雑なのか

toC 向けアプリの認証は、基本的に「ログインしているか」「していないか」の 2 択です。しかし、B2B SaaS（エンタープライズ SaaS） では、以下のような複雑な要件が発生します。

• 複数のユーザー属性: 「管理者」「一般ユーザー」だけでなく、「部署」「役職」「業種」ごとに異なる権限
• 会社間データ分離: 競合他社のデータを絶対に閲覧させない（データ漏洩リスク）
• ページ単位・API 単位のアクセス制御: 「この画面は営業部のみ」「この API は管理者のみ」
• 動的な権限変更: ユーザーの役職変更に応じて、権限を即座に変更

本記事では、私が開発した経済産業大臣賞受賞プロダクト（木材流通業界 DX の B2B SaaS）で実装した、8 種類のユーザー属性を管理する認証・認可設計を、AWS Cognito + Terraform を使った実装例とともに公開します。

アーキテクチャ概要：3 層防御モデル

B2B SaaS の認証・認可は、3 層の防御で設計します。

Layer 1: 認証層（Authentication）
  ↓ AWS Cognito User Pools
  ↓ JWTトークン発行

Layer 2: 認可層（Authorization）
  ↓ カスタム属性による権限チェック
  ↓ ページ単位・API単位のアクセス制御

Layer 3: データ層（Data Access Control）
  ↓ 会社IDによるRow-Level Security（RLS）
  ↓ 署名付きURL（Pre-signed URL）

この 3 層すべてを突破しない限り、データにアクセスできません。

Layer 1: 認証層 - AWS Cognito User Pools の設計

要件：8 種類のユーザー属性を管理

木材流通業界では、以下の 8 種類のユーザー属性が存在します。

Terraform による Cognito User Pool 設計

# AWS Cognito User Pool
resource "aws_cognito_user_pool" "main" {
  name = "lumber-saas-user-pool"

  # パスワードポリシー（エンタープライズ基準）
  password_policy {
    minimum_length    = 12
    require_lowercase = true
    require_uppercase = true
    require_numbers   = true
    require_symbols   = true
  }

  # MFA（多要素認証）必須化
  mfa_configuration = "OPTIONAL" # 本番環境ではON推奨

  # アカウントロックアウト設定（ブルートフォース攻撃対策）
  account_recovery_setting {
    recovery_mechanism {
      name     = "verified_email"
      priority = 1
    }
  }

  # カスタム属性：ユーザー属性（8種類）
  schema {
    name                = "user_type"
    attribute_data_type = "String"
    mutable             = true # 後で変更可能

    string_attribute_constraints {
      min_length = 1
      max_length = 50
    }
  }

  # カスタム属性：会社ID（データ分離用）
  schema {
    name                = "company_id"
    attribute_data_type = "String"
    mutable             = false # 作成後変更不可（セキュリティ）

    string_attribute_constraints {
      min_length = 1
      max_length = 100
    }
  }

  # カスタム属性：権限リスト（カンマ区切り）
  schema {
    name                = "permissions"
    attribute_data_type = "String"
    mutable             = true

    string_attribute_constraints {
      max_length = 2048
    }
  }

  # Eメール検証必須
  auto_verified_attributes = ["email"]

  # タグ
  tags = {
    Environment = var.environment
    Project     = "lumber-saas"
  }
}

# App Client（フロントエンド用）
resource "aws_cognito_user_pool_client" "app" {
  name         = "lumber-saas-app-client"
  user_pool_id = aws_cognito_user_pool.main.id

  # OAuth 2.0 フロー
  allowed_oauth_flows_user_pool_client = true
  allowed_oauth_flows                  = ["code", "implicit"]
  allowed_oauth_scopes                 = ["email", "openid", "profile"]

  # トークン有効期限
  access_token_validity  = 1  # 1時間
  id_token_validity      = 1  # 1時間
  refresh_token_validity = 30 # 30日

  # コールバックURL（本番環境では実際のドメイン）
  callback_urls = [
    "https://lumber-saas.example.com/callback",
    "http://localhost:3000/callback" # 開発環境
  ]

  logout_urls = [
    "https://lumber-saas.example.com/logout",
    "http://localhost:3000/logout"
  ]

  # PKCE必須化（セキュリティ強化）
  prevent_user_existence_errors = "ENABLED"
}

# Cognito Identity Pool（AWS リソースアクセス用）
resource "aws_cognito_identity_pool" "main" {
  identity_pool_name               = "lumber_saas_identity_pool"
  allow_unauthenticated_identities = false

  cognito_identity_providers {
    client_id               = aws_cognito_user_pool_client.app.id
    provider_name           = aws_cognito_user_pool.main.endpoint
    server_side_token_check = true
  }
}

# IAM Role（認証済みユーザー用）
resource "aws_iam_role" "authenticated" {
  name = "cognito_authenticated_role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect = "Allow"
      Principal = {
        Federated = "cognito-identity.amazonaws.com"
      }
      Action = "sts:AssumeRoleWithWebIdentity"
      Condition = {
        StringEquals = {
          "cognito-identity.amazonaws.com:aud" = aws_cognito_identity_pool.main.id
        }
        "ForAnyValue:StringLike" = {
          "cognito-identity.amazonaws.com:amr" = "authenticated"
        }
      }
    }]
  })
}

# S3アクセスポリシー（会社IDごとのフォルダ分離）
resource "aws_iam_role_policy" "authenticated_s3_policy" {
  name = "authenticated_s3_policy"
  role = aws_iam_role.authenticated.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect = "Allow"
      Action = [
        "s3:GetObject",
        "s3:PutObject"
      ]
      Resource = [
        # ユーザーは自社のフォルダのみアクセス可能
        "arn:aws:s3:::lumber-saas-bucket/$${cognito-identity.amazonaws.com:sub}/*"
      ]
    }]
  })
}

Layer 2: 認可層 - ページ単位・API 単位のアクセス制御

フロントエンド：Next.js Middleware での認可チェック

// middleware.ts
import { NextResponse } from "next/server";
import type { NextRequest } from "next/server";
import { jwtVerify } from "jose";

// ページごとの必要権限定義
const PAGE_PERMISSIONS: Record<string, string[]> = {
  "/dashboard": [], // 全ユーザー
  "/inventory": ["view_inventory"],
  "/orders/create": ["create_order"],
  "/users/manage": ["manage_users"], // 管理者のみ
  "/settings/company": ["manage_company"],
};

export async function middleware(request: NextRequest) {
  const token = request.cookies.get("access_token")?.value;

  if (!token) {
    return NextResponse.redirect(new URL("/login", request.url));
  }

  try {
    // JWTトークン検証（AWS Cognito公開鍵使用）
    const { payload } = await jwtVerify(
      token,
      // AWS Cognito公開鍵（JWKSから取得）
      await getPublicKey()
    );

    const userType = payload["custom:user_type"] as string;
    const permissions = (payload["custom:permissions"] as string).split(",");

    // ページ単位の権限チェック
    const requiredPermissions =
      PAGE_PERMISSIONS[request.nextUrl.pathname] || [];
    const hasPermission = requiredPermissions.every((perm) =>
      permissions.includes(perm)
    );

    if (!hasPermission) {
      return NextResponse.redirect(new URL("/403", request.url));
    }

    // 権限OK
    return NextResponse.next();
  } catch (error) {
    console.error("Token verification failed:", error);
    return NextResponse.redirect(new URL("/login", request.url));
  }
}

export const config = {
  matcher: [
    "/dashboard/:path*",
    "/inventory/:path*",
    "/orders/:path*",
    "/users/:path*",
    "/settings/:path*",
  ],
};

バックエンド：Flask デコレータによる API 認可

from functools import wraps
from flask import request, jsonify
import jwt
import requests

# AWS Cognito JWKS URL
COGNITO_JWKS_URL = "https://cognito-idp.ap-northeast-1.amazonaws.com/{user_pool_id}/.well-known/jwks.json"

def get_public_key(token):
    """AWS CognitoのJWKSから公開鍵を取得"""
    jwks = requests.get(COGNITO_JWKS_URL).json()
    # JWTヘッダーから kid を取得し、対応する公開鍵を返す
    # （実装省略、jose ライブラリ等を使用）
    return public_key

def require_permissions(*required_permissions):
    """権限チェックデコレータ"""
    def decorator(f):
        @wraps(f)
        def decorated_function(*args, **kwargs):
            token = request.headers.get('Authorization', '').replace('Bearer ', '')

            if not token:
                return jsonify({'error': 'No token provided'}), 401

            try:
                # トークン検証
                public_key = get_public_key(token)
                payload = jwt.decode(token, public_key, algorithms=['RS256'])

                user_type = payload.get('custom:user_type')
                permissions = payload.get('custom:permissions', '').split(',')
                company_id = payload.get('custom:company_id')

                # 権限チェック
                for perm in required_permissions:
                    if perm not in permissions:
                        return jsonify({'error': f'Permission denied: {perm}'}), 403

                # リクエストコンテキストにユーザー情報を保存
                request.user = {
                    'user_type': user_type,
                    'permissions': permissions,
                    'company_id': company_id,
                }

                return f(*args, **kwargs)
            except jwt.ExpiredSignatureError:
                return jsonify({'error': 'Token expired'}), 401
            except jwt.InvalidTokenError as e:
                return jsonify({'error': f'Invalid token: {str(e)}'}), 401

        return decorated_function
    return decorator

# API エンドポイント例
@app.route('/api/inventory', methods=['GET'])
@require_permissions('view_inventory')
def get_inventory():
    """在庫取得（view_inventory 権限必須）"""
    company_id = request.user['company_id']

    # 自社の在庫のみ取得（他社データは取得不可）
    inventory = Inventory.query.filter_by(company_id=company_id).all()

    return jsonify([inv.to_dict() for inv in inventory])

@app.route('/api/orders', methods=['POST'])
@require_permissions('create_order')
def create_order():
    """発注作成（create_order 権限必須）"""
    company_id = request.user['company_id']
    data = request.json

    # 発注データに会社IDを強制付与（改ざん防止）
    order = Order(
        company_id=company_id,  # 必ずリクエスト元の会社ID
        product_id=data['product_id'],
        quantity=data['quantity'],
    )
    db.session.add(order)
    db.session.commit()

    return jsonify(order.to_dict()), 201

Layer 3: データ層 - Row-Level Security（RLS）

PostgreSQL RLS によるデータ分離

-- テーブル作成（在庫テーブル）
CREATE TABLE inventory (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    company_id UUID NOT NULL, -- 会社ID（必須）
    product_name VARCHAR(255) NOT NULL,
    stock INT NOT NULL,
    price DECIMAL(10, 2) NOT NULL,
    created_at TIMESTAMP DEFAULT NOW()
);

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

-- ポリシー: ユーザーは自社のデータのみ閲覧可能
CREATE POLICY company_isolation_policy ON inventory
    FOR SELECT
    USING (company_id = current_setting('app.current_company_id')::UUID);

-- ポリシー: ユーザーは自社のデータのみ更新可能
CREATE POLICY company_isolation_update_policy ON inventory
    FOR UPDATE
    USING (company_id = current_setting('app.current_company_id')::UUID);

-- ポリシー: ユーザーは自社のデータのみ削除可能
CREATE POLICY company_isolation_delete_policy ON inventory
    FOR DELETE
    USING (company_id = current_setting('app.current_company_id')::UUID);

SQLAlchemy（Python）での実装

from sqlalchemy import event
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "connect")
def set_company_id(dbapi_conn, connection_record):
    """接続ごとに company_id を設定"""
    # Flaskリクエストコンテキストから company_id を取得
    if hasattr(request, 'user'):
        company_id = request.user['company_id']
        cursor = dbapi_conn.cursor()
        cursor.execute(f"SET app.current_company_id = '{company_id}'")
        cursor.close()

# クエリ実行（自動的に自社データのみ取得）
inventory = Inventory.query.all()  # company_id が自動フィルタされる

署名付き URL（Pre-signed URL）による画像・PDF 保護

課題：S3 オブジェクトへの直接アクセス防止

画像、PDF、Excel などのファイルは、S3 に保存されます。しかし、S3 の URL を直接公開すると、URL を知っている人は誰でもアクセス可能です。

解決策：署名付き URL（Pre-signed URL）

import boto3
from botocore.exceptions import ClientError

s3_client = boto3.client('s3', region_name='ap-northeast-1')

def generate_presigned_url(bucket_name, object_key, expiration=3600):
    """
    署名付きURL生成（有効期限付き）

    Args:
        bucket_name: S3バケット名
        object_key: S3オブジェクトキー
        expiration: 有効期限（秒）デフォルト1時間

    Returns:
        str: 署名付きURL
    """
    try:
        url = s3_client.generate_presigned_url(
            'get_object',
            Params={'Bucket': bucket_name, 'Key': object_key},
            ExpiresIn=expiration
        )
        return url
    except ClientError as e:
        print(f"Error generating presigned URL: {e}")
        return None

# API エンドポイント
@app.route('/api/invoices/<invoice_id>/pdf', methods=['GET'])
@require_permissions('view_invoice')
def get_invoice_pdf(invoice_id):
    """請求書PDF取得（署名付きURL）"""
    company_id = request.user['company_id']

    # 請求書の所有権確認
    invoice = Invoice.query.filter_by(id=invoice_id, company_id=company_id).first()
    if not invoice:
        return jsonify({'error': 'Invoice not found'}), 404

    # 署名付きURL生成（1時間有効）
    s3_key = f"invoices/{company_id}/{invoice_id}.pdf"
    presigned_url = generate_presigned_url('lumber-saas-bucket', s3_key, expiration=3600)

    return jsonify({'url': presigned_url})

セキュリティのメリット:

• 有効期限付き: 1 時間後に自動的に URL が無効化
• 所有権確認: 自社の請求書のみアクセス可能
• S3 バケットはプライベート: 直接アクセス不可

まとめ：エンタープライズ SaaS の認証・認可設計の 3 原則

1. 3 層防御を徹底せよ - 認証層、認可層、データ層の全てで権限チェック
2. 会社間データ分離を絶対視せよ - RLS、company_id の強制付与で、他社データ漏洩を完全防止
3. IaC で再現性を確保せよ - Terraform でインフラ全体をコード化、セキュリティ設定の属人化を防ぐ

B2B SaaS の認証・認可は、toC 向けアプリとは次元が異なる複雑さです。しかし、AWS Cognito + Terraform + RLS の組み合わせにより、エンタープライズレベルのセキュリティを実現できます。

あなたの B2B SaaS も実現可能です

複雑な認証・認可設計、AWS Cognito 設定、Terraform IaC に悩んでいる方、私がサポートします。要件定義から設計、実装まで、ワンストップで対応可能です。

無料技術相談（30 分） を実施していますので、お気軽にご連絡ください。

お問い合わせはこちら