AWS Cognito で企業SSOを実装する完全ガイド：SAML/OIDC連携（Azure AD・Okta・Google）とUSER_AUTH選択式認証

[ユーザー] → [SaaSフロント(Next.js/Amplify)]
   → マネージドログインへリダイレクト
   → [Cognito User Pool] が SAMLRequest 生成
   → [顧客IdP: Azure AD] で認証 + SAML Assertion
   → [Cognito] が ACS (/saml2/idpresponse) で受領 → ユーザー自動作成 → JWT発行
   → [SaaSバック] が JWT を aws-jwt-verify で検証 → テナントID抽出 → API認可

# --- User Pool 本体（料金プランを明示）---
resource "aws_cognito_user_pool" "main" {
  name = "my-saas-user-pool"

  # 2024/11 導入: 機能プラン。アクセストークン拡張・パスワードレスを使うなら ESSENTIALS 以上。
  user_pool_tier = "ESSENTIALS"

  # マルチテナント用カスタム属性（IdPから受け取るテナントID）
  schema {
    name                = "tenant_id"
    attribute_data_type = "String"
    mutable             = true # IdP マッピング属性は mutable 必須
    string_attribute_constraints {
      min_length = 1
      max_length = 256
    }
  }

  auto_verified_attributes = ["email"]

  # SSO主体でもローカルユーザー用にポリシーは定義しておく
  password_policy {
    minimum_length    = 12
    require_lowercase = true
    require_uppercase = true
    require_numbers   = true
    require_symbols   = true
  }

  account_recovery_setting {
    recovery_mechanism {
      name     = "verified_email"
      priority = 1
    }
  }
}

# --- SAML Identity Provider（Azure AD）---
resource "aws_cognito_identity_provider" "azure_ad_saml" {
  user_pool_id  = aws_cognito_user_pool.main.id
  provider_name = "AzureAD" # supported_identity_providers と app側の指定で使う名前
  provider_type = "SAML"

  provider_details = {
    MetadataURL = "https://login.microsoftonline.com/<tenant-id>/federationmetadata/2007-06/federationmetadata.xml"
    # シングルログアウト（IdP起点・SP起点）を有効化する場合
    IDPSignout = "true"
    # Cognito が送る SAMLRequest の署名アルゴリズム（推奨）
    RequestSigningAlgorithm = "rsa-sha256"
  }

  # SAMLクレーム → Cognito属性 のマッピング
  attribute_mapping = {
    email               = "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/emailaddress"
    given_name          = "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/givenname"
    family_name         = "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/surname"
    "custom:tenant_id"  = "http://schemas.microsoft.com/identity/claims/tenantid"
  }
}

# --- ドメイン（マネージドログイン）---
resource "aws_cognito_user_pool_domain" "main" {
  domain       = "my-saas-domain"
  user_pool_id = aws_cognito_user_pool.main.id
}

# --- App Client ---
resource "aws_cognito_user_pool_client" "web_client" {
  name         = "web-client"
  user_pool_id = aws_cognito_user_pool.main.id

  allowed_oauth_flows                  = ["code"]            # 認可コードフロー（推奨）
  allowed_oauth_scopes                 = ["openid", "email", "profile"]
  allowed_oauth_flows_user_pool_client = true
  callback_urls                        = ["https://my-app.com/auth/callback"]
  logout_urls                          = ["https://my-app.com/logout"]
  supported_identity_providers         = ["AzureAD", "COGNITO"]

  # 認証フロー（後述の USER_AUTH を使う場合は ALLOW_USER_AUTH を追加）
  explicit_auth_flows = ["ALLOW_REFRESH_TOKEN_AUTH", "ALLOW_USER_SRP_AUTH"]

  # トークン有効期限（単位を必ず明示する。既定は hours だが意図を明確に）
  access_token_validity  = 1
  id_token_validity      = 1
  refresh_token_validity = 30
  token_validity_units {
    access_token  = "hours"
    id_token      = "hours"
    refresh_token = "days"
  }
}

resource "aws_cognito_identity_provider" "google_oidc" {
  user_pool_id  = aws_cognito_user_pool.main.id
  provider_name = "Google"
  provider_type = "Google"

  provider_details = {
    client_id        = var.google_oauth_client_id
    client_secret    = var.google_oauth_client_secret
    authorize_scopes = "openid email profile" # openid は必須
  }

  attribute_mapping = {
    email       = "email"
    given_name  = "given_name"
    family_name = "family_name"
    username    = "sub"
  }
}

resource "aws_cognito_identity_provider" "okta_oidc" {
  user_pool_id  = aws_cognito_user_pool.main.id
  provider_name = "Okta"
  provider_type = "OIDC"

  provider_details = {
    client_id                 = var.okta_client_id
    client_secret             = var.okta_client_secret
    authorize_scopes          = "openid email profile" # openid 必須
    oidc_issuer               = "https://<your-org>.okta.com"
    attributes_request_method = "GET" # userInfo の取得メソッド
  }

  attribute_mapping = {
    email       = "email"
    given_name  = "given_name"
    family_name = "family_name"
    username    = "sub"
  }
}

InitiateAuth(CUSTOM_AUTH)
        │
        ▼
DefineAuthChallenge ──「次は何のチャレンジ？／合否は？」を判断（司令塔）
        │  challengeName = CUSTOM_CHALLENGE
        ▼
CreateAuthChallenge ──チャレンジの中身を生成（OTPコード生成＆送信など）
        │  publicChallengeParameters（クライアントへ）/ privateChallengeParameters（答え）
        ▼
（クライアントが RespondToAuthChallenge で ANSWER を送信）
        │
        ▼
VerifyAuthChallengeResponse ──答え合わせ（answerCorrect: true/false）
        │
        ▼
DefineAuthChallenge ──結果を見て「トークン発行」or「もう一度」or「失敗」を決定

# define_auth_challenge.py
def lambda_handler(event, context):
    session = event["request"]["session"]

    if len(session) == 0:
        # 初回: 独自チャレンジを提示
        event["response"]["challengeName"] = "CUSTOM_CHALLENGE"
        event["response"]["issueTokens"] = False
        event["response"]["failAuthentication"] = False
    elif (
        len(session) == 1
        and session[0]["challengeName"] == "CUSTOM_CHALLENGE"
        and session[0]["challengeResult"] is True
    ):
        # 1回正解したらトークン発行
        event["response"]["issueTokens"] = True
        event["response"]["failAuthentication"] = False
    else:
        # それ以外は失敗（無限ループ・総当たり防止）
        event["response"]["issueTokens"] = False
        event["response"]["failAuthentication"] = True

    return event

# create_auth_challenge.py
import secrets
import boto3

ses = boto3.client("ses")

def lambda_handler(event, context):
    if event["request"]["challengeName"] != "CUSTOM_CHALLENGE":
        return event

    # 例: 6桁のワンタイムコードを生成して送信（独自のCAPTCHA等でも同様）
    code = f"{secrets.randbelow(1_000_000):06d}"
    email = event["request"]["userAttributes"]["email"]

    ses.send_email(
        Source="no-reply@my-saas.com",
        Destination={"ToAddresses": [email]},
        Message={
            "Subject": {"Data": "ログイン確認コード"},
            "Body": {"Text": {"Data": f"確認コード: {code}（5分間有効）"}},
        },
    )

    # クライアントに見せてよい情報のみ public へ
    event["response"]["publicChallengeParameters"] = {"medium": "EMAIL"}
    # 答えは private（応答には絶対に含めない）
    event["response"]["privateChallengeParameters"] = {"answer": code}
    event["response"]["challengeMetadata"] = "EMAIL_ONE_TIME_CODE"
    return event

# verify_auth_challenge.py
def lambda_handler(event, context):
    expected = event["request"]["privateChallengeParameters"]["answer"]
    provided = event["request"]["challengeAnswer"]
    event["response"]["answerCorrect"] = secrets_equal(expected, provided)
    return event

def secrets_equal(a: str, b: str) -> bool:
    # タイミング攻撃を避けるため定数時間比較
    import hmac
    return hmac.compare_digest(a, b)

resource "aws_cognito_user_pool" "main" {
  # ... 既出の設定 ...
  user_pool_tier = "ESSENTIALS"

  # 第一認証要素として許可する方式（WEB_AUTHN は他要素と併用が必須）
  sign_in_policy {
    allowed_first_auth_factors = ["PASSWORD", "EMAIL_OTP", "WEB_AUTHN"]
  }
}

resource "aws_cognito_user_pool_client" "web_client" {
  # ... 既出の設定 ...
  explicit_auth_flows = [
    "ALLOW_USER_AUTH",            # 選択式サインインを有効化
    "ALLOW_REFRESH_TOKEN_AUTH",
  ]
}

// パスワードレス・サインイン（メールOTPを選好）
import { signIn, confirmSignIn } from "aws-amplify/auth";

async function startPasswordless(username: string) {
  const { nextStep } = await signIn({
    username,
    options: { authFlowType: "USER_AUTH", preferredChallenge: "EMAIL_OTP" },
  });

  if (nextStep.signInStep === "CONFIRM_SIGN_IN_WITH_EMAIL_CODE") {
    // ユーザーがメールで受け取ったコードを入力 → confirmSignIn で送信
  }
}

async function submitOtp(code: string) {
  const { isSignedIn } = await confirmSignIn({ challengeResponse: code });
  return isSignedIn;
}

// パスキー（WebAuthn）の登録（ログイン後に呼ぶ）
import { associateWebAuthnCredential } from "aws-amplify/auth";

async function registerPasskey() {
  // ブラウザのWebAuthn APIをAmplifyが内部で起動し、資格情報をCognitoに登録
  await associateWebAuthnCredential();
}

// LoginPage.tsx（Amplify v6 / Next.js）
import { signInWithRedirect, signIn } from "aws-amplify/auth";
import { useState } from "react";

type Tenant = { id: string; name: string; idp: "AzureAD" | "Google" | "Okta" | null };

export function LoginPage() {
  const [email, setEmail] = useState("");
  const [tenant, setTenant] = useState<Tenant | null>(null);

  // メールドメインからテナントを判定
  async function detectTenant(value: string) {
    const res = await fetch("/api/detect-tenant", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ email: value }),
    });
    setTenant((await res.json()).tenant);
  }

  // 企業SSO（フェデレーション）へ
  async function loginWithSSO() {
    if (!tenant?.idp) return;
    await signInWithRedirect({ provider: { custom: tenant.idp } });
  }

  // 個人ユーザーはパスワードレスへ
  async function loginPasswordless() {
    await signIn({
      username: email,
      options: { authFlowType: "USER_AUTH", preferredChallenge: "EMAIL_OTP" },
    });
  }

  return (
    <form
      onSubmit={(e) => {
        e.preventDefault();
        tenant?.idp ? loginWithSSO() : loginPasswordless();
      }}
    >
      <label htmlFor="email">メールアドレス</label>
      <input
        id="email"
        type="email"
        autoComplete="email webauthn"
        value={email}
        onChange={(e) => setEmail(e.target.value)}
        onBlur={() => detectTenant(email)}
      />
      <button type="submit">
        {tenant?.idp ? `${tenant.name} のSSOでログイン` : "確認コードでログイン"}
      </button>
    </form>
  );
}

# api/detect_tenant.py
from fastapi import APIRouter
from pydantic import BaseModel, EmailStr
import boto3

router = APIRouter()
tenants = boto3.resource("dynamodb").Table("tenants")

class Req(BaseModel):
    email: EmailStr

@router.post("/api/detect-tenant")
async def detect_tenant(req: Req):
    domain = req.email.split("@")[1]
    res = tenants.query(
        IndexName="domain-index",
        KeyConditionExpression="#d = :d",
        ExpressionAttributeNames={"#d": "domain"},
        ExpressionAttributeValues={":d": domain},
    )
    if not res["Items"]:
        return {"tenant": None}  # 未登録 → 個人ユーザー扱い
    t = res["Items"][0]
    return {"tenant": {"id": t["tenant_id"], "name": t["company_name"], "idp": t.get("idp")}}

resource "aws_cognito_user_pool" "main" {
  # ... 既出の設定 ...
  lambda_config {
    pre_token_generation_config {
      lambda_arn     = aws_lambda_function.pre_token.arn
      lambda_version = "V2_0" # アクセストークンに載せるなら V2_0 以上
    }
  }
}

# pre_token_generation.py （V2_0）
def lambda_handler(event, context):
    attrs = event["request"]["userAttributes"]
    tenant_id = attrs.get("custom:tenant_id", "")

    event["response"]["claimsAndScopeOverrideDetails"] = {
        # アクセストークンにテナントIDとスコープを注入
        "accessTokenGeneration": {
            "claimsToAddOrOverride": {"tenant_id": tenant_id},
            "scopesToAdd": [f"tenant/{tenant_id}"],
        },
        # 必要ならIDトークンにも
        "idTokenGeneration": {
            "claimsToAddOrOverride": {"tenant_id": tenant_id},
        },
        # cognito:groups を上書きしてテナント別ロールを反映することも可能
        # "groupOverrideDetails": {"groupsToOverride": [f"tenant-{tenant_id}-member"]},
    }
    return event

import { CognitoJwtVerifier } from "aws-jwt-verify";

// アクセストークン用検証器（client_id と token_use を自動チェック）
const verifier = CognitoJwtVerifier.create({
  userPoolId: process.env.COGNITO_USER_POOL_ID!,
  tokenUse: "access",
  clientId: process.env.COGNITO_CLIENT_ID!,
});

export async function authorize(authHeader: string | null) {
  const token = authHeader?.replace(/^Bearer\s+/i, "");
  if (!token) throw new Response("Unauthorized", { status: 401 });

  try {
    const payload = await verifier.verify(token); // 署名 + iss + exp + client_id + token_use
    return {
      userId: payload.sub,
      tenantId: payload["tenant_id"] as string | undefined, // Pre Token Gen で注入
      groups: (payload["cognito:groups"] as string[] | undefined) ?? [],
    };
  } catch {
    throw new Response("Invalid token", { status: 401 });
  }
}

# middleware/auth.py
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from jwt import PyJWKClient
import os

security = HTTPBearer()
REGION = os.environ["COGNITO_REGION"]
POOL_ID = os.environ["COGNITO_USER_POOL_ID"]
CLIENT_ID = os.environ["COGNITO_CLIENT_ID"]
ISSUER = f"https://cognito-idp.{REGION}.amazonaws.com/{POOL_ID}"
jwks = PyJWKClient(f"{ISSUER}/.well-known/jwks.json")

def verify_access_token(creds: HTTPAuthorizationCredentials = Security(security)) -> dict:
    token = creds.credentials
    try:
        key = jwks.get_signing_key_from_jwt(token).key
        payload = jwt.decode(token, key, algorithms=["RS256"], issuer=ISSUER)
        # アクセストークンは aud を持たないため client_id / token_use を手動検証
        if payload.get("token_use") != "access" or payload.get("client_id") != CLIENT_ID:
            raise jwt.InvalidTokenError("token_use / client_id mismatch")
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError as e:
        raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

Invalid SAML response: Signature validation failed

InvalidParameterException: User does not have a valid email

import { Amplify } from "aws-amplify";

Amplify.configure({
  Auth: {
    Cognito: {
      userPoolId: "ap-northeast-1_XXXXXXXXX",
      userPoolClientId: "xxxxxxxxxxxxxxxxxxxx",
      loginWith: {
        oauth: {
          domain: "my-saas-domain.auth.ap-northeast-1.amazoncognito.com",
          scopes: ["openid", "email", "profile"],
          redirectSignIn: ["https://my-app.com/auth/callback"], // 末尾スラッシュ含め完全一致
          redirectSignOut: ["https://my-app.com/logout"],
          responseType: "code",
        },
      },
    },
  },
});