# AWS Cognito カスタム認証フロー実装ガイド：CUSTOM_AUTH チャレンジで OTP/パスワードレス、PIN は PBKDF2 で安全に保管する

> CognitoのCUSTOM_AUTHチャレンジ（Define/Create/VerifyのLambdaトリガー）でOTP・パスワードレス・LINE認証を実装し、カードPINはPBKDF2-HMAC（高反復・CSPRNGソルト・定数時間比較）で安全に保管する実装ガイド。確認後フックやログのマスキングまで実コードで解説。

- 公開日: 2026-06-24
- 著者: 友田 陽大
- タグ: AWS, Cognito, 認証, セキュリティ, Python
- URL: https://tomodahinata.com/blog/aws-cognito-custom-authentication-pin-pbkdf2-passwordless-guide

## 要点

- CUSTOM_AUTHは認証フローの制御、PBKDF2は秘密の保管で、レイヤーが違う別問題として分離する
- Define/Create/Verifyの3トリガーで責務を分け、試行回数の打ち切りはDefine、定数時間照合はVerifyが担う
- MFAが欲しいだけなら標準の組み込みMFAで足り、CUSTOM_AUTHは組み込みにない独自チャレンジ専用と考える
- PINはPBKDF2-HMACで一方向ハッシュ化し、CSPRNGソルト・反復回数を同梱、検証はhmac.compare_digestで定数時間比較する
- 4桁PINは鍵空間が小さく、ハッシュ強度より試行回数のレート制限が第一防御になる

---

「ユーザー名とパスワードでログインしたい」——それだけなら、Cognito の標準フローで終わりです。設定すれば動きます。

ところが現場の要件は、たいていそこからはみ出します。**パスワードを使わせたくない（パスワードレス）。メールや SMS で届く OTP（ワンタイムコード）で入らせたい。LINE のアカウントと連携させたい。店頭端末では 4 桁の PIN で素早く認証させたい。** これらは「ユーザー名＋パスワード」という Cognito の組み込みモデルには、そのままでは載りません。

このとき使うのが **CUSTOM_AUTH（カスタム認証チャレンジ）** です。Cognito が用意した 3 つの Lambda トリガー（**DefineAuthChallenge / CreateAuthChallenge / VerifyAuthChallengeResponse**）に、「どんな問いを、どう出し、どう検証するか」を自分で書く仕組みです。

そしてもう一つ、混同してはいけない別問題があります。**PIN やパスワードのような「秘密そのもの」をどう保管するか**です。CUSTOM_AUTH は認証フローの制御であって秘密の保管ではありません。秘密の保管は、**PBKDF2-HMAC による一方向ハッシュ化**という別レイヤーの話です。

この記事は、私が環境・カーボンクレジット／地域通貨のマルチテナント決済プラットフォーム（顧客・加盟店・管理・店頭端末の 4 面、AWS サーバーレス、本番二重課金 0 件）で実装した、**Cognito カスタム認証フロー**と**カード PIN の安全な保管**の設計判断を、実コードで解説します。

> **この記事のルール**：仕様・パラメータ名・推奨値の出典は **AWS Cognito 公式ドキュメントおよび OWASP 公式ドキュメント（2026年6月時点）** です。トリガーのイベント形状や反復回数は改定され得るため、本番投入前に必ず[末尾の公式ドキュメント](#参考公式ドキュメント)で最新を確認してください。**秘密情報（PIN・OTP・トークン類）は必ずハッシュ化・マスキングし、ログには一切出しません。** コードは実運用想定で整えていますが、シークレットは環境変数／Secrets Manager 前提です（ハードコード厳禁）。

この記事は「カスタム認証フローの設計と実装」と「秘密の保管」に絞ります。役割分担として、関連する 3 本を先に置いておきます。

- **どの認証方式を組むべきか／複雑な要件の全体設計**：[複雑な認証要件のCognito設計](/blog/aws-cognito-complex-authentication-design)
- **SAML/OIDC によるエンタープライズ SSO**：[SAML/OIDCエンタープライズSSO](/blog/aws-cognito-saml-oidc-enterprise-sso)
- **発行後のトークン（JWT）を API 側でどう検証するか**：[Cognito JWT(RS256)検証](/blog/aws-cognito-jwt-rs256-verification-jwks-security-guide)

**設計・SSO・トークン検証はそちらへ。本記事はカスタム認証フローと秘密の保管に集中します。**

---

## 0. メンタルモデル：3 トリガーで「チャレンジを自分で定義する」

最初に頭の地図を作ります。標準のユーザー名＋パスワードでは足りない要件（OTP・パスワードレス・LINE 連携・PIN）に対して、Cognito は「チャレンジ（問いと答え）を自分で定義する」余地を与えてくれます。それが 3 つの Lambda トリガーです。

| トリガー | 役割（一言で） | 入力で見るもの | 出力で決めるもの |
| --- | --- | --- | --- |
| **DefineAuthChallenge** | フロー制御：次に何のチャレンジを出すか | `event.request.session`（これまでの結果） | `challengeName` / `issueTokens` / `failAuthentication` |
| **CreateAuthChallenge** | チャレンジ内容：OTP を生成・送信する | `event.request.challengeName` / `session` | `publicChallengeParameters` / `privateChallengeParameters` / `challengeMetadata` |
| **VerifyAuthChallengeResponse** | 回答の検証：答えが正しいか | `privateChallengeParameters` / `challengeAnswer` | `answerCorrect` |

公式の整理がわかりやすいので引きます。**Define は「チャレンジの順序」を管理し、Create は「チャレンジの中身」を管理し、Verify は「既知の正解とアプリが送ってきた回答を突き合わせる」**。この 3 つが鎖のようにつながり、「完全に自分の設計どおりの認証メカニズム」を作ります（公式 "user-pool-lambda-challenge"）。

そして、これと**独立した別問題**が「秘密の保管」です。

- **CUSTOM_AUTH**：その場限りの OTP を「出して・送って・照合する」フロー。OTP は一度使ったら捨てる。
- **PBKDF2 ハッシュ**：PIN やパスワードという「長期に保持する秘密」を、**復元不可能な一方向ハッシュ**にして DB に置く。検証は「同じ手順でハッシュ化して定数時間比較」する。

この 2 つを混ぜないことが設計の出発点です。OTP は CUSTOM_AUTH の `privateChallengeParameters` に短時間だけ持たせ、PIN は DB に PBKDF2 ハッシュとして長期保管する。レイヤーが違います（SRP：関心の分離）。

---

## 1. いつ標準認証で足り、いつ CUSTOM_AUTH が必要か

カスタム認証は強力ですが、**アプリ側にもログイン UI と複数回のチャレンジ応答ロジックを書く必要があり、マネージドログインでは処理できません**（公式が明言しています）。つまり「開発者の追加コスト」を払う判断です。安易に選ぶと負債になります。まずは要件が標準で足りるかを判定してください。

| 要件 | 標準フロー（USER_SRP_AUTH 等） | CUSTOM_AUTH |
| --- | --- | --- |
| ユーザー名＋パスワード | ✅ そのまま | 不要 |
| MFA（SMS / TOTP / メール OTP） | ✅ 組み込み MFA で足りる | 不要（標準 MFA を使う） |
| パスワードレス（OTP のみでサインイン） | ❌ | ✅ 適任 |
| LINE / 独自 IdP の OTP・確認コード | ❌ | ✅ 適任 |
| 4桁 PIN・セキュリティ質問・CAPTCHA を「認証の一部」に | ❌ | ✅ 適任 |
| ハードウェアキー・生体・外部 API での独自検証 | ❌ | ✅ 適任 |

判断は KISS と YAGNI で。**MFA が欲しいだけなら、Cognito の組み込み MFA（`SMS_MFA` / `SOFTWARE_TOKEN_MFA` / `EMAIL_OTP`）で十分**です。わざわざ 3 トリガーを書く必要はありません。`USER_SRP_AUTH` で SRP 認証を済ませると、MFA が設定済みのユーザーには Cognito が自動で `SMS_MFA` などを続けてくれます。

CUSTOM_AUTH が真に必要なのは、**「Cognito の組み込みにない問い」を認証の一部にしたいとき**だけです。OTP を自前のチャネル（LINE、独自メール基盤）で送りたい、PIN を認証フローに組み込みたい、外部の生体認証 API の結果を「正解」として扱いたい——こういう要件で初めて元が取れます。

> 私の決済プラットフォームでは、店頭端末・加盟店・顧客とチャネルが分かれており、**LINE／メール OTP 等の多様なサインインを 1 つの仕組みで吸収する**必要がありました。ここが CUSTOM_AUTH の出番です。逆に、普通のメール＋パスワード口は標準フローのままにして、複雑さを必要な面だけに閉じ込めました（複雑さの局所化）。

---

## 2. フローの全体像：InitiateAuth から IssueTokens まで

CUSTOM_AUTH は、`InitiateAuth`（または `AdminInitiateAuth`）で始まり、`RespondToAuthChallenge` を繰り返し、最後に **DefineAuthChallenge が `issueTokens: true` を返した瞬間にトークンが発行されて成功**します。チャレンジの応答が「次のチャレンジ」になることもあり、必要なだけ往復します（公式）。

パスワードレスで「PIN だけ」「OTP だけ」で入らせたい場合は、SRP のパスワード検証を挟まず、最初から CUSTOM_AUTH チャレンジに入れます。公式いわく、**パスワード検証から始めたくなければ、`AuthParameters` の `CHALLENGE_NAME` を `CUSTOM_CHALLENGE` にして `InitiateAuth` を呼べばよい**とのことです。

```jsonc
// パスワードレス開始：いきなり CUSTOM_CHALLENGE から
{
  "AuthFlow": "CUSTOM_AUTH",
  "ClientId": "1example23456789",
  "AuthParameters": {
    "USERNAME": "testuser",
    "CHALLENGE_NAME": "CUSTOM_CHALLENGE"
    // SECRET_HASH はアプリクライアントにシークレットがある場合に付与
  }
}
```

一方、「パスワードを検証してから、さらに OTP も要求する」多要素にしたい場合は、SRP を先に通します。公式が示す順序はこうです（"SRP authentication in custom challenge flows"）。

1. アプリが `InitiateAuth` を `CHALLENGE_NAME: SRP_A`・`SRP_A`・`USERNAME` で呼ぶ。
2. Cognito が DefineAuthChallenge を、`session` に `challengeName: SRP_A` / `challengeResult: true` を入れて呼ぶ。
3. Lambda は `challengeName: PASSWORD_VERIFIER`、`issueTokens: false`、`failAuthentication: false` を返す。
4. パスワード検証が成功すると、Cognito は `challengeName: PASSWORD_VERIFIER` / `challengeResult: true` の `session` で再度呼ぶ。
5. Lambda は `challengeName: CUSTOM_CHALLENGE` を返し、独自チャレンジに入る。
6. チャレンジのループは、すべて答え終わるまで繰り返す。

ここで覚えておくべきは、**`session` は配列で、これまで提示・解決されたチャレンジが時系列で積まれる**ことです。`session[0]` が最初のチャレンジ。DefineAuthChallenge は毎回この配列を見て「次に何を出すか」を決めます。これがフロー制御の心臓部です。

---

## 3. DefineAuthChallenge：フローを制御する

最初に書くのは Define です。**「これまでの session を見て、次の challengeName を返す。全部終わっていれば issueTokens=true、失敗させたければ failAuthentication=true」**——責務はこれだけです（SRP）。

公式が定義するイベント形状（抜粋）は次のとおりです。

- `event.request.session`：`ChallengeResult` の配列。各要素は `challengeName` / `challengeResult`（成功なら `true`）/ `challengeMetadata`（`CUSTOM_CHALLENGE` のときの自分用の名前）。
- `event.request.userNotFound`：`PreventUserExistenceErrors` を `ENABLED` にしていると、存在しないユーザーで `true` になる。
- `event.response.challengeName` / `issueTokens` / `failAuthentication`：次に何をするか。

ここで **`userNotFound` の扱いが地味に重要**です。公式は「ユーザーが存在してもしなくても同じ挙動・同じ遅延を保ち、呼び出し側が違いを検知できないようにせよ」と推奨しています。存在しないユーザーで早期に `failAuthentication=true` を返すと、**応答時間や挙動の差からユーザーの存在を推測される**（ユーザー列挙攻撃）からです。

```python
"""DefineAuthChallenge: 認証フローのオーケストレーション。
ここでは『パスワードレスで OTP 1回』のシンプル構成を制御する。
- まだ何も解いていなければ -> CUSTOM_CHALLENGE を出す
- OTP に正解したら -> トークン発行
- 規定回数失敗したら -> 認証を打ち切る
"""
MAX_CHALLENGE_ATTEMPTS = 3  # OTP の試行上限（Verify と整合させる）


def handler(event, context):
    req = event["request"]
    res = event["response"]
    session = req.get("session") or []

    # 1) まだチャレンジを出していない -> 最初の CUSTOM_CHALLENGE
    if len(session) == 0:
        res["issueTokens"] = False
        res["failAuthentication"] = False
        res["challengeName"] = "CUSTOM_CHALLENGE"
        return event

    last = session[-1]

    # 2) 直近の CUSTOM_CHALLENGE に正解 -> トークン発行
    if last["challengeName"] == "CUSTOM_CHALLENGE" and last["challengeResult"] is True:
        res["issueTokens"] = True
        res["failAuthentication"] = False
        return event

    # 3) 失敗回数が上限に達した -> 打ち切り（OTP 総当たりの抑止）
    failed = sum(
        1 for c in session
        if c["challengeName"] == "CUSTOM_CHALLENGE" and c["challengeResult"] is False
    )
    if failed >= MAX_CHALLENGE_ATTEMPTS:
        res["issueTokens"] = False
        res["failAuthentication"] = True
        return event

    # 4) まだ猶予がある -> もう一度 CUSTOM_CHALLENGE（同じ OTP を再入力させる）
    res["issueTokens"] = False
    res["failAuthentication"] = False
    res["challengeName"] = "CUSTOM_CHALLENGE"
    return event
```

公式の Node.js サンプルは「`session.length` と各要素の `challengeName` を見て分岐する」形でした。私は**長さに直接依存せず「直近の結果」と「失敗回数の集計」で判定**しています。理由は ETC（Easy To Change）：将来チャレンジを 1 つ増やしたとき、`length === 3` のようなマジックナンバーを全部ずらすのは壊れやすいからです。意味（最後に正解したか／何回失敗したか）で書けば、段数が変わっても影響が局所で済みます。

> **重要**：トークンを発行する前に、公式は「`challengeName` を必ずチェックし、期待した値であることを確認せよ」と明記しています。`issueTokens=true` を返す分岐では、**直近のチャレンジが本当に意図したもの**かを必ず確認してください。ここを雑にすると、想定外のチャレンジ結果でトークンが出る穴になり得ます。

---

## 4. CreateAuthChallenge：OTP を生成して安全に送る

Define が `CUSTOM_CHALLENGE` を指定すると、Cognito は CreateAuthChallenge を呼びます。ここが**チャレンジの中身**——OTP を生成し、ユーザーへ送り、**正解を `privateChallengeParameters` に隠して**Cognito に預ける場所です。

公式が定義するイベント形状（抜粋）：

- `event.request.challengeName`：次のチャレンジ名（Define が決めた値）。
- `event.response.publicChallengeParameters`：**クライアントに見せてよい**情報（「OTP をメールに送りました」等のヒント）。
- `event.response.privateChallengeParameters`：**Verify トリガーだけが使う**情報。公式の言葉で言えば「`publicChallengeParameters` が利用者に提示する“問い”を、`privateChallengeParameters` がその問いの“正解”を含む」。
- `event.response.challengeMetadata`：このカスタムチャレンジに付ける自分用の名前。

### 4.1 OTP は CSPRNG で作る

OTP の生成で最初に間違えてはいけないのが乱数源です。**`random` モジュールは暗号用途に使ってはいけません**（予測可能）。Python なら `secrets` を使います。

```python
"""CreateAuthChallenge: OTP を生成し、メール/LINE で送り、
正解を privateChallengeParameters に隠す。
- 乱数は secrets（CSPRNG）で生成
- 平文 OTP は publicChallengeParameters に絶対入れない
- OTP・宛先はログに出さない
"""
import json
import os
import secrets
import time

OTP_TTL_SECONDS = 300  # 有効期限 5 分


def generate_otp() -> str:
    # 000000〜999999 を CSPRNG で。先頭ゼロ込みの 6 桁固定。
    return f"{secrets.randbelow(1_000_000):06d}"


def handler(event, context):
    req = event["request"]
    res = event["response"]

    if req["challengeName"] != "CUSTOM_CHALLENGE":
        return event  # 自分の担当外は素通し（防御的）

    # 既存セッションに OTP があれば再送せず使い回す（同一チャレンジの再表示時）
    existing = _existing_otp(req.get("session") or [])
    otp = existing or generate_otp()

    if existing is None:
        _deliver_otp(req["userAttributes"], otp)  # 送信は副作用として分離（SRP）

    # クライアントに見せてよい情報だけ public に
    res["publicChallengeParameters"] = {
        "deliveryMedium": "EMAIL",          # 「メールに送った」程度の UI ヒント
        "maskedDestination": _mask_email(req["userAttributes"].get("email", "")),
    }
    # 正解と期限は private に（Verify だけが読む）
    res["privateChallengeParameters"] = {
        "otp": otp,
        "expiresAt": str(int(time.time()) + OTP_TTL_SECONDS),
    }
    res["challengeMetadata"] = "EMAIL_OTP"
    return event


def _existing_otp(session: list) -> str | None:
    # challengeMetadata を手がかりに、直近の自分のチャレンジを判定して引き継ぐ
    for c in reversed(session):
        if c.get("challengeMetadata") == "EMAIL_OTP":
            # 注意: session の要素から private は読めない設計なので、
            # 再送制御は外部ストア(後述)で行うのが本筋。ここは概念図。
            return None
    return None
```

ここに 2 つの設計上の注意があります。

1. **平文 OTP を `publicChallengeParameters` に入れない**。`public` はクライアントに渡る＝攻撃者の手元にも届き得ます。OTP は `private` にだけ。`public` には「メールに送った」「宛先末尾は ***@example.com」程度のヒントだけを置きます。
2. **`privateChallengeParameters` は便利だが過信しない**。`private` は Verify に届きますが、Cognito のチャレンジ往復に紐づくため、**OTP の有効期限・再送回数のような“状態”は、DynamoDB のような外部ストアで TTL 管理する**ほうが堅牢です（後述の落とし穴）。

### 4.2 配送はチャネルごとに分離する（SRP）

OTP の「送信」は、生成や検証とは別の関心です。メール（SES）・SMS（SNS）・LINE（Messaging API）でチャネルが違うだけなので、配送だけを差し替え可能にしておきます。

```python
"""OTP 配送: チャネルを抽象化。生成・検証から切り離す（ETC）。"""
import boto3

_ses = boto3.client("ses")


def _deliver_otp(user_attrs: dict, otp: str) -> None:
    email = user_attrs.get("email")
    phone = user_attrs.get("phone_number")
    # LINE 連携ユーザーなら line_user_id を見て Messaging API に切替、等
    if email:
        _send_email_otp(email, otp)
    elif phone:
        _send_sms_otp(phone, otp)
    # ※ ここで otp も email/phone も print/log しない（PII + 秘密）


def _send_email_otp(to: str, otp: str) -> None:
    _ses.send_email(
        Source="no-reply@example.com",
        Destination={"ToAddresses": [to]},
        Message={
            "Subject": {"Data": "ログイン用ワンタイムコード"},
            "Body": {"Text": {"Data": f"認証コード: {otp}（5分間有効）"}},
        },
    )
```

> **IAM は最小権限で**。CreateAuthChallenge の実行ロールに必要なのは `ses:SendEmail` 等の**配送権限だけ**。決済 DB や鍵への権限は与えません。私のプラットフォームでは **IAM をスタック単位で最小権限に分離**し、認証 Lambda が決済リソースに触れない設計にしました（最小権限の原則）。

---

## 5. VerifyAuthChallengeResponse：回答を定数時間で検証する

Create が出した問いにユーザーが答えると、Cognito は Verify を呼びます。**`event.request.privateChallengeParameters`（正解）と `event.request.challengeAnswer`（ユーザーの回答）を比べ、`event.response.answerCorrect` に `true`/`false` を返す**——それだけです（公式）。

公式の Node.js サンプルは `privateChallengeParameters.answer === challengeAnswer` という**素朴な `===` 比較**でした。これはサンプルとしては正しいのですが、**秘密の比較に通常の `==` / `===` を使うとタイミング攻撃の余地が残ります**。文字列比較は「最初に違うバイトで打ち切る」ため、一致するプレフィックスが長いほど処理が長引き、その差から正解を 1 文字ずつ絞り込まれ得ます。

OTP の照合は**定数時間比較**で行います。Python なら `hmac.compare_digest` です。

```python
"""VerifyAuthChallengeResponse: OTP を定数時間で照合し、期限・形式も検査する。
- hmac.compare_digest で一致比較（タイミング攻撃対策）
- 期限切れ・形式不正は即 false
- OTP も回答もログに出さない
"""
import hmac
import time


def handler(event, context):
    req = event["request"]
    res = event["response"]

    private = req.get("privateChallengeParameters") or {}
    expected = private.get("otp", "")
    answer = (req.get("challengeAnswer") or "").strip()

    res["answerCorrect"] = _verify_otp(expected, answer, private.get("expiresAt"))
    return event


def _verify_otp(expected: str, answer: str, expires_at: str | None) -> bool:
    # 形式チェック（6桁数字）。境界で弾く＝総当たり面積を減らす
    if len(answer) != 6 or not answer.isdigit():
        return False
    # 有効期限（Create で privateChallengeParameters に入れた値）
    if expires_at is not None and int(time.time()) > int(expires_at):
        return False
    if not expected:
        return False
    # 定数時間比較。長さ差で早期 return しないよう encode してから比較
    return hmac.compare_digest(expected.encode("utf-8"), answer.encode("utf-8"))
```

ポイントは 3 つです。

1. **定数時間比較**：`hmac.compare_digest` は入力長に依存せず一定時間で比較します。OTP・PIN・トークンの照合は必ずこれを使います。
2. **境界での形式検証**：6 桁数字でなければ即 `false`。外部入力（`challengeAnswer`）は素通しせず、境界で検証します（入力検証）。
3. **有効期限**：期限切れ OTP は無効。期限は Create が `privateChallengeParameters.expiresAt` に入れた値で判定します。

> **試行回数の上限は Verify だけでは閉じない**。Verify は 1 回の照合の正否を返すだけです。「3 回失敗で打ち切り」は、§3 の Define が `session` 内の失敗回数を数えて `failAuthentication=true` を返すことで成立します。**ロックアウトの最終判断は Define、その場の照合は Verify**——責務を取り違えないでください。

---

## 6. サインアップ／確認後フック：post-confirmation トリガー

カスタム認証の「サインイン」とは別に、**サインアップの確認後**にやりたい初期化があります。たとえば「確認済みユーザーに決済プロファイル行を作る」「初回の PIN 登録レコードを用意する」など。これを担うのが **PostConfirmation トリガー**です。

公式によれば、PostConfirmation は **`ConfirmSignUp` / `AdminConfirmSignUp` / `ConfirmForgotPassword`**（およびマネージドログインでのサインアップ／パスワードリセット確認）で発火します。`event.request` には確定済みユーザーの `userAttributes` と `clientMetadata` が入り、**レスポンスに返すべき追加情報はありません**（`"response": {}`）。

```python
"""PostConfirmation: サインアップ確認後の初期化。
- triggerSource でサインアップ確認とパスワードリセット確認を区別
- 副作用は冪等に（同じ確認が2回来ても二重作成しない）
- 例外で握りつぶさず、ユーザー体験を壊さない範囲で記録
"""
import boto3

_ddb = boto3.resource("dynamodb")
_profiles = _ddb.Table("payment_user_profiles")


def handler(event, context):
    source = event.get("triggerSource")  # 例: PostConfirmation_ConfirmSignUp
    attrs = event["request"]["userAttributes"]
    sub = attrs["sub"]  # Cognito の不変ユーザーID（メールではなく sub をキーに）

    if source == "PostConfirmation_ConfirmSignUp":
        _ensure_profile(sub)  # 冪等な作成
    elif source == "PostConfirmation_ConfirmForgotPassword":
        _mark_password_reset(sub)  # リセット完了を記録（PIN 再登録を促す等）

    return event  # response は空のまま返す


def _ensure_profile(sub: str) -> None:
    # 条件付き書き込みで「既にあれば作らない」= 冪等（再実行・重複イベント耐性）
    try:
        _profiles.put_item(
            Item={"user_sub": sub, "status": "active"},
            ConditionExpression="attribute_not_exists(user_sub)",
        )
    except _ddb.meta.client.exceptions.ConditionalCheckFailedException:
        pass  # 既存ならスキップ（正常系）
```

設計上の勘所が 2 つあります。

1. **冪等にする**：確認イベントは再送・重複し得ます。`ConditionExpression="attribute_not_exists(...)"` で「あれば作らない」を保証し、二重作成を防ぎます（信頼性）。決済プラットフォームで「二重作成・二重課金 0 件」を守る感覚と同じです。
2. **`sub` をキーにする**：メールアドレスは変わり得ます。Cognito の不変 ID である `sub`（`userAttributes.sub`）を主キーに使い、メールはあくまで連絡先として扱います。

> **pre-signup との使い分け**：登録を「許可するか／自動確認するか」を制御したいなら PreSignUp、確認**後**の初期化なら PostConfirmation です。本記事の主題（カスタム認証フロー）では、確認後にユーザー周辺データを整える PostConfirmation が中心になります。なお post-confirmation は重い同期処理でユーザー体験をブロックしないよう、外部 I/O は最小に、失敗は適切に記録します。

---

## 7. 秘密の保管：カード PIN は PBKDF2-HMAC で一方向ハッシュ化する

ここからが「別問題」です。OTP はその場限りで捨てますが、**PIN やパスワードは長期に保持する秘密**です。これを**平文で保存するのは論外**。可逆暗号で保存するのも危険です（鍵が漏れれば全件平文に戻る）。**正解は一方向ハッシュ化**——`hashlib.pbkdf2_hmac` を使い、検証時は「同じ手順でハッシュ化して定数時間比較」します。

### 7.1 アルゴリズムの選択：PBKDF2 か、Argon2id か

まず選定です。OWASP のパスワード保管チートシートは、優先順位を明確にしています。

| アルゴリズム | OWASP の位置づけ | 推奨パラメータ（公式値） | いつ選ぶか |
| --- | --- | --- | --- |
| **Argon2id** | 第一推奨 | `m=19456 (19 MiB), t=2, p=1` | メモリハードを使えるなら第一候補 |
| **scrypt** | 次点 | — | Argon2id が使えない環境 |
| **bcrypt** | レガシー向け | 入力上限 **72 バイト** | 既存システムの維持のみ |
| **PBKDF2** | **FIPS-140 が必要なら最優先** | 下表（HMAC 別） | NIST/FIPS 準拠が要件のとき |

OWASP は PBKDF2 について「**NIST に推奨され、FIPS-140 検証済み実装がある**ため、それらが要件のときは優先すべき」としています。つまり、**純粋な強度だけなら Argon2id ですが、規制・コンプライアンス（FIPS-140 準拠）が要件なら PBKDF2 が正解**です。

決済・金融まわりは FIPS-140 準拠を求められることがあり、かつ Python 標準ライブラリ（`hashlib`）だけで完結して依存を増やさない利点もあります。私のプラットフォームでは **PBKDF2-HMAC を採用**しました。これは「最強だから」ではなく「要件（標準ライブラリ・実装の枯れ具合・コンプライアンス）に最も合うから」という選定です（YAGNI／コスト効率：余計な依存を増やさない）。

### 7.2 反復回数：OWASP の公式値

PBKDF2 の強度は反復回数で決まります。OWASP の現行推奨は HMAC のハッシュ関数ごとに異なります。

| PBKDF2 の HMAC | OWASP 推奨反復回数 | 備考 |
| --- | --- | --- |
| **PBKDF2-HMAC-SHA256** | **600,000 回** | 一般的な推奨 |
| **PBKDF2-HMAC-SHA512** | **220,000 回** | — |
| PBKDF2-HMAC-SHA1 | 1,400,000 回 | **レガシーのみ** |

> 私のプラットフォームを構築した時点では **10 万回反復** を採用していました。これは当時の要件・端末性能とのトレードオフで決めた値です。**反復回数は年々引き上げる前提**であり、新規実装では上表の OWASP 現行値（SHA256 なら 600,000 回）を基準に、自分の本番ハードでのハッシュ所要時間を計測して決めてください。反復回数・ソルト長・アルゴリズムは**保存フォーマットに同梱**し、後から段階的に引き上げられるようにします（ETC）。

### 7.3 実装：ソルト同梱・バージョン付きの保存フォーマット

PIN ハッシュは「ソルト・反復回数・アルゴリズム」を**ハッシュと一緒に保存**します。そうしないと、後でパラメータを変えたときに既存ハッシュを検証できなくなります。

```python
"""PIN/パスワードを PBKDF2-HMAC で一方向ハッシュ化して保管する。
- ソルトは 32 バイト CSPRNG（os.urandom / secrets）で毎回生成
- ソルト・反復回数・アルゴリズムを文字列に同梱（後から引き上げ可能に）
- 検証は hmac.compare_digest で定数時間比較
- 平文 PIN は受け取った関数の外に出さない/ログにも出さない
"""
import base64
import hashlib
import hmac
import secrets

# 新規はこの値を基準に（OWASP: SHA256 は 600_000）。本番ハードで計測して調整。
PBKDF2_ALGO = "sha256"
PBKDF2_ITERATIONS = 600_000
SALT_BYTES = 32          # 32 バイト CSPRNG ソルト
DERIVED_KEY_LEN = 32     # 出力長


def hash_pin(pin: str) -> str:
    """平文 PIN -> 保存用文字列。形式: pbkdf2$<algo>$<iter>$<salt_b64>$<hash_b64>"""
    salt = secrets.token_bytes(SALT_BYTES)            # 毎回ユニーク（CSPRNG）
    dk = hashlib.pbkdf2_hmac(
        PBKDF2_ALGO, pin.encode("utf-8"), salt, PBKDF2_ITERATIONS, DERIVED_KEY_LEN
    )
    return "$".join([
        "pbkdf2",
        PBKDF2_ALGO,
        str(PBKDF2_ITERATIONS),
        base64.b64encode(salt).decode("ascii"),
        base64.b64encode(dk).decode("ascii"),
    ])


def verify_pin(pin: str, stored: str) -> bool:
    """平文 PIN と保存済みハッシュを定数時間で照合。"""
    try:
        scheme, algo, iter_s, salt_b64, hash_b64 = stored.split("$")
    except ValueError:
        return False
    if scheme != "pbkdf2":
        return False

    salt = base64.b64decode(salt_b64)
    expected = base64.b64decode(hash_b64)
    # 保存時と同じパラメータで再導出（だからこそ同梱が必須）
    candidate = hashlib.pbkdf2_hmac(
        algo, pin.encode("utf-8"), salt, int(iter_s), len(expected)
    )
    # 定数時間比較。True/False の分岐を比較結果のみに依存させる
    return hmac.compare_digest(candidate, expected)
```

この実装で押さえている点を整理します。

- **ソルトは 32 バイト CSPRNG・パスワードごとにユニーク**。`secrets.token_bytes(32)` で生成し、ハッシュと一緒に保存します。ソルトがないと、同じ PIN が同じハッシュになり、レインボーテーブルや「同じ PIN を使う人の一括特定」が成立します。**ソルトは秘密ではないので同梱して構いません**——役割は「同一入力を別ハッシュにする」ことです。
- **反復回数・アルゴリズムを同梱**。`pbkdf2$sha256$600000$...$...` の形にしておけば、将来 SHA256/600k → より強い設定へ移行するとき、ログイン成功時に**その場で再ハッシュ**して段階移行できます（ETC）。
- **定数時間比較**。検証の最後は必ず `hmac.compare_digest`。文字列の `==` は使いません。

> **4 桁 PIN の現実**：4 桁 PIN は鍵空間が 1 万通りしかなく、ハッシュをいくら強くしてもオフライン総当たりには弱い。**だから PIN は「ハッシュ強度」だけでなく「試行回数のレート制限」と「端末／コンテキスト束縛」で守る**のが本質です。PIN ハッシュは漏洩時の被害を遅らせる最後の砦であって、第一の防御は「サーバー側で試行回数を絞ること」。ここを誤解すると、強い PBKDF2 を入れて安心してしまいます。

---

## 8. セキュリティの落とし穴（必読チェックリスト）

実装が動くことと、安全であることは別です。CUSTOM_AUTH と秘密保管で、実際に踏みやすい落とし穴を列挙します。

- **OTP の試行回数を絞る**：Define の `session` で失敗回数を数え、3〜5 回で `failAuthentication=true`。さらに「OTP 発行レート（同一ユーザーへの再送間隔）」も外部ストアで制限する。これがないと OTP 総当たり・送信スパムの的になります。
- **OTP に有効期限を付ける**：Create で `expiresAt` を `privateChallengeParameters` に入れ、Verify で必ずチェック。期限なし OTP は「いつまでも使える鍵」と同じです。
- **平文 OTP / PIN / トークンを `publicChallengeParameters` に入れない**：`public` はクライアントに渡る＝攻撃者にも届く。正解は必ず `private` にだけ。
- **比較は定数時間で**：OTP・PIN・トークン・署名の照合は `hmac.compare_digest`。`==` / `===` はタイミング攻撃の入口になります。
- **ソルトは必須・CSPRNG・毎回ユニーク**：ソルトなしハッシュはレインボーテーブルで一括逆引きされ得る。ソルトは秘密でなくてよいが、ハッシュごとに違うことが必須。
- **平文保存・可逆暗号保存をしない**：PIN/パスワードは一方向ハッシュのみ。可逆暗号は鍵漏洩で全件平文化する単一障害点になります。
- **PIN / OTP / トークンをログに出さない**：例外スタックトレースやデバッグログに秘密が漏れがち。**ログはメール・電話をマスクし、PIN・トークン類は一切出力しない**。
- **`PreventUserExistenceErrors` を ENABLED に**：`userNotFound` で挙動・遅延を変えない。ユーザー列挙攻撃を防ぎます。
- **IAM はスタック単位で最小権限**：認証 Lambda の実行ロールには配送権限など必要最小限だけ。決済 DB・鍵管理への権限を渡さない。
- **OTP の再利用を許さない**：一度成功（または失敗で打ち切り）した OTP は無効化。同じコードを使い回せないようにする。

### ログのマスキング：秘密を絶対に出さない

最後に、最も事故が多い「ログ」を潰します。**秘密を持つ値は、ログ出力の前に必ずマスク**します。

```python
"""ログ用マスキング: PII はマスク、秘密は出力しない。
構造化ログにはユーザーID(sub)・チャレンジ名・成否などメタデータだけ残す。
"""
import re

_SECRET_KEYS = {"otp", "pin", "password", "token", "challengeAnswer",
                "privateChallengeParameters", "secret_hash"}


def mask_email(value: str) -> str:
    # a***@example.com の形に
    if "@" not in value:
        return "***"
    local, _, domain = value.partition("@")
    head = local[:1] if local else ""
    return f"{head}***@{domain}"


def mask_phone(value: str) -> str:
    return re.sub(r"\d(?=\d{2})", "*", value)  # 末尾2桁だけ残す


def safe_log_payload(event_request: dict) -> dict:
    """ログに出してよい最小メタデータだけを抽出する。"""
    attrs = event_request.get("userAttributes", {})
    return {
        "sub": attrs.get("sub"),                       # 不変ID（PIIではない識別子）
        "email": mask_email(attrs.get("email", "")),   # マスク
        "phone": mask_phone(attrs.get("phone_number", "")),
        "challengeName": event_request.get("challengeName"),
        # ↑ otp / pin / token / challengeAnswer は意図的に含めない
    }
```

ルールはシンプルです。**秘密キー（OTP・PIN・トークン・回答・private パラメータ）はログ用 payload に最初から入れない。** 「マスクし忘れる」のではなく「そもそも入れない」設計（許可リスト方式）にすれば、事故が構造的に起きません。

---

## 9. まとめ：チートシート

最後に、迷ったときの早見表です。

**いつ CUSTOM_AUTH を使うか**

- MFA が欲しいだけ → **標準の組み込み MFA**（CUSTOM_AUTH 不要）。
- パスワードレス・OTP を自前チャネルで・PIN を認証に組み込む・外部 API で独自検証 → **CUSTOM_AUTH**。

**3 トリガーの責務**

- **Define** = フロー制御。`session` を見て次の `challengeName` / `issueTokens` / `failAuthentication` を返す。**試行回数の打ち切りはここ**。
- **Create** = 中身。OTP を CSPRNG で生成・送信し、正解を `privateChallengeParameters` に、ヒントだけ `publicChallengeParameters` に、名前を `challengeMetadata` に。
- **Verify** = 照合。`challengeAnswer` と `privateChallengeParameters` を **`hmac.compare_digest`（定数時間）** で比較し `answerCorrect` を返す。期限・形式も検査。
- **PostConfirmation** = 確認後の初期化。`triggerSource` で確認/リセットを区別し、`sub` をキーに**冪等**に。

**秘密の保管**

- PIN/パスワードは **PBKDF2-HMAC で一方向ハッシュ化**（FIPS-140 要件なら最優先、純強度なら Argon2id）。
- **反復回数は OWASP 現行値**（PBKDF2-HMAC-SHA256＝600,000 回）を基準に本番ハードで計測。
- **32 バイト CSPRNG ソルトを毎回ユニークに、ハッシュと同梱**。アルゴリズム・反復回数も同梱して後から引き上げ可能に。
- **検証は定数時間比較**。平文保存・可逆暗号・ログ出力は禁止。
- 4 桁 PIN は鍵空間が小さいので、**ハッシュ強度より試行回数のレート制限が第一防御**。

CUSTOM_AUTH は「認証フローを自分で設計する自由」を与えてくれますが、その自由は**自分でセキュリティを担保する責任**とセットです。OTP の試行制限、定数時間比較、ソルト、ログのマスキング、最小権限 IAM——この基本を外すと、柔軟さがそのまま穴になります。

私は環境分野のサーバーレス決済プラットフォームで、**Cognito のカスタム認証（サインアップ／確認後フック／CUSTOM_AUTH チャレンジ）で LINE・メール OTP 等の多様なサインインを実現**し、**カード PIN を PBKDF2-HMAC（CSPRNG ソルト・定数時間比較）で安全に保管**しました。入力は境界で検証し、IAM はスタック単位で最小権限に分離し、ログはメール・電話をマスクして PIN・トークンは一切出さない——この設計で**本番二重課金 0 件**を守りました。

**「自社の認証要件（パスワードレス・OTP・PIN・外部 IdP 連携）をどう Cognito に落とし込み、秘密をどう安全に保管するか」——要件整理から実装・運用まで一気通貫で伴走します。** 認証は事故ると信用に直結する領域です。設計段階からでも、お気軽にご相談ください。

前後関係として、認証方式の選定・全体設計は[複雑な認証要件のCognito設計](/blog/aws-cognito-complex-authentication-design)、社外 IdP との SSO は[SAML/OIDCエンタープライズSSO](/blog/aws-cognito-saml-oidc-enterprise-sso)、発行後トークンの検証は[Cognito JWT(RS256)検証](/blog/aws-cognito-jwt-rs256-verification-jwks-security-guide)にまとめています。あわせてどうぞ。

---

### 参考（公式ドキュメント）

- [Custom authentication challenge Lambda triggers（AWS Cognito）](https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-lambda-challenge.html) — CUSTOM_AUTH フロー全体・SRP 連携・InitiateAuth/RespondToAuthChallenge の往復
- [Define Auth challenge Lambda trigger（AWS Cognito）](https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-lambda-define-auth-challenge.html) — `session` / `challengeName` / `issueTokens` / `failAuthentication`
- [Create Auth challenge Lambda trigger（AWS Cognito）](https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-lambda-create-auth-challenge.html) — `publicChallengeParameters` / `privateChallengeParameters` / `challengeMetadata`
- [Verify Auth challenge response Lambda trigger（AWS Cognito）](https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-lambda-verify-auth-challenge-response.html) — `challengeAnswer` / `answerCorrect`
- [Post confirmation Lambda trigger（AWS Cognito）](https://docs.aws.amazon.com/cognito/latest/developerguide/user-pool-lambda-post-confirmation.html) — `triggerSource` / `ConfirmSignUp` / `ConfirmForgotPassword`
- [Password Storage Cheat Sheet（OWASP）](https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html) — PBKDF2 反復回数・ソルト・アルゴリズム優先順位・FIPS-140