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 設計
# 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原則
- 3層防御を徹底せよ - 認証層、認可層、データ層の全てで権限チェック
- 会社間データ分離を絶対視せよ - RLS、company_id の強制付与で、他社データ漏洩を完全防止
- IaCで再現性を確保せよ - Terraform でインフラ全体をコード化、セキュリティ設定の属人化を防ぐ
B2B SaaSの認証・認可は、toC向けアプリとは次元が異なる複雑さです。しかし、AWS Cognito + Terraform + RLS の組み合わせにより、エンタープライズレベルのセキュリティを実現できます。
あなたのB2B SaaSも実現可能です
複雑な認証・認可設計、AWS Cognito設定、Terraform IaCに悩んでいる方、私がサポートします。要件定義から設計、実装まで、ワンストップで対応可能です。
無料技術相談（30分） を実施していますので、お気軽にご連絡ください。