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

> B2B SaaSで必須となる、複雑なユーザー属性ごとの認証・認可をAWS Cognitoで実現する方法を解説。8種類のユーザー属性、ページ単位・API単位のアクセス制御、署名付きURL、Terraform IaCによる完全自動化まで、実践的な設計パターンを公開。

- 公開日: 2025-01-07
- 著者: 友田 陽大
- タグ: AWS Cognito, Terraform, 認証設計, B2B SaaS, セキュリティ, IaC, アクセス制御
- URL: https://tomodahinata.com/blog/aws-cognito-complex-authentication-design
- カテゴリ: 認証・認可
- 総合ガイド: https://tomodahinata.com/blog/auth-platform-selection-2026-cognito-auth0-clerk-supabase

## 要点

- B2B SaaSの認証・認可は認証層・認可層・データ層の3層防御で設計し、すべてを突破しない限りデータに届かせない
- 8種類の業種属性はCognitoのカスタム属性で持ち、company_idは作成後変更不可（mutable=false）にしてデータ分離を担保する
- 認可はページ単位・API単位で権限チェックし、発注などの書き込みには会社IDをサーバ側で強制付与して改ざんを防ぐ
- データ層はPostgreSQL RLSでcompany_idによる行レベル分離を徹底し、他社データ漏洩を構造的に封じる
- S3のファイルは署名付きURL（有効期限＋所有権確認）で保護し、バケットはプライベートに保つ

---

# 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 種類のユーザー属性が存在します。

| ユーザー属性   | 説明       | 主な権限             |
| -------------- | ---------- | -------------------- |
| `lumber_mill`  | 製材所     | 在庫管理、発注受注   |
| `market`       | 市場       | 全在庫閲覧、価格設定 |
| `manufacturer` | メーカー   | 製品登録、在庫管理   |
| `precut`       | プレカット | 加工依頼、納品管理   |
| `builder`      | 工務店     | 発注、見積依頼       |
| `wholesaler`   | 問屋       | 卸売価格設定、発注   |
| `forestry`     | 林業       | 原木登録、市場出荷   |
| `other`        | その他     | 閲覧のみ             |

### Terraform による Cognito User Pool 設計

```hcl
# 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 での認可チェック

```typescript
// 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 認可

```python
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 によるデータ分離

```sql
-- テーブル作成（在庫テーブル）
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）での実装

```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）

```python
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 分）** を実施していますので、お気軽にご連絡ください。

[お問い合わせはこちら](/contact)