PUSHPASSKEY_API_KEY=ここにAPIキーを記載してください
PUSHPASSKEY_TEST_MODE=true
PUSHPASSKEY_TEST_USER_KEY=pushpasskey_test_user
PUSHPASSKEY_REQUEST_TIMEOUT_SECONDS=15
SESSION_SECRET=十分に長いランダムな文字列へ変更してください
ENVIRONMENT=development

import logging
import os
from typing import Any
from urllib.parse import urlparse

import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import FileResponse, JSONResponse
from starlette.middleware.sessions import SessionMiddleware

load_dotenv()

API_BASE = "https://auth.jintec.com"
COMPLETED_STATUSES = {"success", "failed", "expired", "cancelled"}
logger = logging.getLogger("pushpasskey")

app = FastAPI(docs_url=None, redoc_url=None)
app.add_middleware(
    SessionMiddleware,
    secret_key=os.environ["SESSION_SECRET"],
    session_cookie="pushpasskey.sid",
    same_site="lax",
    https_only=os.getenv("ENVIRONMENT") == "production",
    max_age=30 * 60,
)


@app.get("/pushpasskey/test", include_in_schema=False)
async def test_page() -> FileResponse:
    return FileResponse("app/static/pushpasskey-test.html")


@app.get("/pushpasskey/status", include_in_schema=False)
async def status(request: Request) -> JSONResponse:
    return JSONResponse({
        "ok": True,
        "authenticated_user_key": request.session.get(
            "pushpasskey_authenticated_user_key", ""
        ),
    })


@app.post("/pushpasskey/endpoint", include_in_schema=False)
async def endpoint(request: Request) -> JSONResponse:
    try:
        verify_same_origin(request)
        body = await request.json()
        action = str(body.get("action", "")).strip()

        if action == "regist_start":
            return await regist_start(request, body)
        if action == "auth_start":
            return await auth_start(request, body)
        if action == "regist_result":
            return await regist_result(request, body)
        if action == "auth_result":
            return await auth_result(request, body)

        return json_error(400, "処理の指定が正しくありません。")
    except ValueError as exc:
        return json_error(400, str(exc))
    except PermissionError as exc:
        return json_error(403, str(exc))
    except (httpx.TimeoutException, httpx.NetworkError):
        logger.exception("Push! Passkey API connection error")
        return json_error(502, "Push! Passkey APIへ接続できませんでした。")
    except Exception:
        logger.exception("Push! Passkey endpoint error")
        return json_error(500, "Push! Passkeyの処理を完了できませんでした。")


# ============================================================
# 本番接続時に変更する箇所 1
# ログイン中のユーザーを一意に識別する文字列を返します。
# ============================================================
def get_logged_in_user_key(request: Request) -> str:
    # 例：return str(request.session.get("user_id", "")).strip()
    if os.getenv("PUSHPASSKEY_TEST_MODE") == "true":
        return str(os.getenv("PUSHPASSKEY_TEST_USER_KEY", "")).strip()
    return ""


# ============================================================
# 本番接続時に変更する箇所 2
# パスキー登録成功後の処理です。
# ============================================================
async def after_regist_success(
    request: Request, result: dict[str, Any]
) -> None:
    request.session["passkey_message"] = "パスキーの登録が完了しました。"
    # 必要に応じて登録済み状態や登録日時をDBへ保存します。


# ============================================================
# 本番接続時に変更する箇所 3
# パスキー認証成功後のログイン処理です。
# ============================================================
async def after_auth_success(
    request: Request, result: dict[str, Any]
) -> None:
    user_key = str(result.get("user_key", "")).strip()
    if not user_key:
        raise RuntimeError("認証ユーザーを確認できません。")

    # 実運用ではuser_keyに一致する有効なユーザーをDBから取得し、
    # 自社サイトのログインセッションを発行します。
    request.session.clear()
    request.session["pushpasskey_authenticated_user_key"] = user_key


async def regist_start(
    request: Request, body: dict[str, Any]
) -> JSONResponse:
    user_key = get_logged_in_user_key(request)
    if not user_key:
        return json_error(401, "ログイン状態を確認できません。")

    result = await call_pushpasskey("/regist_url", {
        "api_key": api_key(),
        "user_key": user_key,
        "return_page_url": return_page_url(request, body),
    })
    if result.get("ok") and result.get("request_id"):
        save_pending_request(
            request, str(result["request_id"]), "regist", user_key
        )
    return JSONResponse(result)


async def auth_start(
    request: Request, body: dict[str, Any]
) -> JSONResponse:
    result = await call_pushpasskey("/auth_url", {
        "api_key": api_key(),
        "return_page_url": return_page_url(request, body),
    })
    if result.get("ok") and result.get("request_id"):
        save_pending_request(request, str(result["request_id"]), "auth")
    return JSONResponse(result)


async def regist_result(
    request: Request, body: dict[str, Any]
) -> JSONResponse:
    request_id = required_request_id(body)
    pending = pending_request(request, request_id, "regist")
    result = await call_pushpasskey("/regist_result", {
        "api_key": api_key(), "request_id": request_id
    })
    if result.get("is_success"):
        if str(result.get("user_key", "")) != str(pending["user_key"]):
            raise RuntimeError("登録ユーザーが一致しません。")
        await after_regist_success(request, result)
    if result.get("is_completed"):
        clear_pending_request(request, request_id)
    return JSONResponse(result)


async def auth_result(
    request: Request, body: dict[str, Any]
) -> JSONResponse:
    request_id = required_request_id(body)
    pending_request(request, request_id, "auth")
    result = await call_pushpasskey("/auth_result", {
        "api_key": api_key(), "request_id": request_id
    })
    status_value = str(result.get("status", "")).strip().lower()
    success = bool(result.get("is_success")) or status_value == "success"
    if success:
        await after_auth_success(request, result)
    if result.get("is_completed") or status_value in COMPLETED_STATUSES:
        clear_pending_request(request, request_id)
    return JSONResponse(result)


async def call_pushpasskey(
    api_path: str, payload: dict[str, Any]
) -> dict[str, Any]:
    if not api_key():
        raise RuntimeError("Push! Passkeyの設定を確認してください。")
    timeout = float(os.getenv("PUSHPASSKEY_REQUEST_TIMEOUT_SECONDS", "15"))
    async with httpx.AsyncClient(
        timeout=timeout, follow_redirects=False
    ) as client:
        response = await client.post(
            f"{API_BASE}{api_path}",
            json=payload,
            headers={"Accept": "application/json"},
        )
    content_type = response.headers.get("content-type", "")
    if "application/json" not in content_type.lower():
        raise RuntimeError("Push! PasskeyからJSON以外の応答が返されました。")
    result = response.json()
    if not isinstance(result, dict):
        raise RuntimeError("Push! Passkeyの応答形式が正しくありません。")
    if not response.is_success:
        result["ok"] = False
    return result


def save_pending_request(
    request: Request, request_id: str, kind: str, user_key: str = ""
) -> None:
    requests = dict(request.session.get("pushpasskey_requests", {}))
    requests[request_id] = {"kind": kind, "user_key": user_key}
    request.session["pushpasskey_requests"] = requests


def pending_request(
    request: Request, request_id: str, kind: str
) -> dict[str, Any]:
    data = request.session.get("pushpasskey_requests", {}).get(request_id)
    if not data or data.get("kind") != kind:
        raise ValueError("処理情報を確認できません。最初からやり直してください。")
    return data


def clear_pending_request(request: Request, request_id: str) -> None:
    requests = dict(request.session.get("pushpasskey_requests", {}))
    requests.pop(request_id, None)
    request.session["pushpasskey_requests"] = requests


def required_request_id(body: dict[str, Any]) -> str:
    request_id = str(body.get("request_id", "")).strip()
    if not request_id:
        raise ValueError("処理情報を確認できません。")
    return request_id


def return_page_url(request: Request, body: dict[str, Any]) -> str:
    value = str(body.get("return_page_url", "")).strip()
    parsed = urlparse(value)
    if parsed.scheme not in {"http", "https"} or parsed.netloc != request.url.netloc:
        return ""
    return value


def verify_same_origin(request: Request) -> None:
    origin = request.headers.get("origin")
    if origin and urlparse(origin).netloc != request.url.netloc:
        raise PermissionError("このページからは処理を開始できません。")


def api_key() -> str:
    return str(os.getenv("PUSHPASSKEY_API_KEY", "")).strip()


def json_error(status_code: int, message: str) -> JSONResponse:
    return JSONResponse(
        {"ok": False, "message": message}, status_code=status_code
    )

<!doctype html>
<html lang="ja">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width,initial-scale=1">
  <title>Push! Passkey動作確認</title>
  <script src="https://auth.jintec.com/js/passkey.js" defer></script>
</head>
<body>
  <h1>Push! Passkey動作確認</h1>
  <p>最初にパスキー登録を行い、続いて認証してください。</p>

  <button type="button" data-pushpasskey="regist"
    data-endpoint="/pushpasskey/endpoint"
    data-success-url="/pushpasskey/test">
    パスキーを登録する
  </button>

  <button type="button" data-pushpasskey="auth"
    data-endpoint="/pushpasskey/endpoint"
    data-success-url="/pushpasskey/test">
    パスキーで認証する
  </button>

  <div id="resultMessage" role="alert" aria-live="polite"></div>
  <p id="sessionStatus">認証状態を確認しています。</p>
  <script>
  fetch('/pushpasskey/status', {credentials: 'same-origin'})
    .then((response) => response.json())
    .then((data) => {
      document.getElementById('sessionStatus').textContent =
        data.authenticated_user_key
          ? `認証したuser_key：${data.authenticated_user_key}`
          : '認証したuser_keyはまだありません。';
    })
    .catch(() => {
      document.getElementById('sessionStatus').textContent =
        '認証状態を取得できませんでした。';
    });
  </script>
</body>
</html>