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

「ユーザー名とパスワードでログインしたい」——それだけなら、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設計
• SAML/OIDC によるエンタープライズ SSO：SAML/OIDCエンタープライズSSO
• 発行後のトークン（JWT）を API 側でどう検証するか：Cognito JWT(RS256)検証

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

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

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

公式の整理がわかりやすいので引きます。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 と複数回のチャレンジ応答ロジックを書く必要があり、マネージドログインでは処理できません（公式が明言しています）。つまり「開発者の追加コスト」を払う判断です。安易に選ぶと負債になります。まずは要件が標準で足りるかを判定してください。

判断は 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 を呼べばよいとのことです。

// パスワードレス開始：いきなり 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 を返すと、応答時間や挙動の差からユーザーの存在を推測される（ユーザー列挙攻撃）からです。

"""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 を使います。

"""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）でチャネルが違うだけなので、配送だけを差し替え可能にしておきます。

"""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 です。

"""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": {}）。

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

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

7.2 反復回数：OWASP の公式値

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

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

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

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

"""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 は無効化。同じコードを使い回せないようにする。

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

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

"""ログ用マスキング: 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設計、社外 IdP との SSO はSAML/OIDCエンタープライズSSO、発行後トークンの検証はCognito JWT(RS256)検証にまとめています。あわせてどうぞ。

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

• Custom authentication challenge Lambda triggers（AWS Cognito） — CUSTOM_AUTH フロー全体・SRP 連携・InitiateAuth/RespondToAuthChallenge の往復
• Define Auth challenge Lambda trigger（AWS Cognito） — session / challengeName / issueTokens / failAuthentication
• Create Auth challenge Lambda trigger（AWS Cognito） — publicChallengeParameters / privateChallengeParameters / challengeMetadata
• Verify Auth challenge response Lambda trigger（AWS Cognito） — challengeAnswer / answerCorrect
• Post confirmation Lambda trigger（AWS Cognito） — triggerSource / ConfirmSignUp / ConfirmForgotPassword
• Password Storage Cheat Sheet（OWASP） — PBKDF2 反復回数・ソルト・アルゴリズム優先順位・FIPS-140