# FastAPI API を保護する

このクイックスタートでは、KeenAuth のアクセストークンで FastAPI API を保護します。Python SDK の `RequireAuth` 依存関係を追加し、1 つのルートにスコープを要求して、SDK のテストキットで API をオフラインでテストします。

この API には 3 つのエンドポイントがあります:

| エンドポイント | 必要条件 |
|---|---|
| `GET /api/public` | 不要 |
| `GET /api/private` | API 用の有効なアクセストークン |
| `GET /api/messages` | `messages:read` スコープを持つ有効なアクセストークン |

## 始める前に

- KeenAuth の環境。その issuer は `https://your-environment.eu.keenauth.app` のような形式です。
- 環境に登録された API（[API を作成する](/docs/api/management/apis/create)）。その `identifier`（例: `https://api.example.com`）が、API が受け入れるトークンの audience です。
- その API に対する `messages:read` スコープ。呼び出し元のアプリケーションに付与されている必要があります。
- Python 3.10 以降。

## 1. SDK をインストールする

```sh
pip install "keenauth-python[fastapi]" uvicorn
```

SDK はまだ公開されていません。公開されるまで、この例では SDK のソースからインストールします。例では、すべてのパッケージを `requirements.txt` で正確なバージョンに固定しています。

## 2. 設定する

API は設定を環境変数からのみ読み込みます:

| 変数 | 必須 | 意味 |
|---|---|---|
| `KEENAUTH_ISSUER` | はい | 環境の issuer |
| `KEENAUTH_AUDIENCE` | はい | API の識別子 |
| `PUBLIC_BASE_URL` | はい | クライアントが呼び出す URL（[プロキシの背後](#プロキシの背後)を参照） |
| `KEENAUTH_WEBHOOK_SECRET` | 推奨 | `token.revoked` を配信する webhook エンドポイントの `idp_whsec_` 署名シークレット（ローテーション中は 2 つをカンマ区切りで指定）。これがないと、API は取り消されたトークンを検知できません |

## 3. ルートを保護する

`create_app` はプロテクタとルートを作成します。`Depends(RequireAuth(protector))` は有効なトークンを要求し、検証済みのクレームを返します。`scopes=[...]` はさらにスコープも要求します。`install(app)` は、エラーを `application/problem+json` レスポンスにし、チャレンジごとに 1 つの `WWW-Authenticate` ヘッダを付与します。

```python
def create_app(
    verifier: AccessTokenVerifier,
    *,
    public_base_url: str,
    revocation: Optional[RevocationChecker] = None,
    webhook_secrets: Sequence[str] = (),
) -> FastAPI:
    """Builds the API. The verifier is a parameter so tests can pass the offline test kit's verifier. With
    webhook_secrets (the idp_whsec_ secrets of your webhook endpoint) the token.revoked webhook is served and feeds
    `revocation`, which must then be a MemoryRevocationCache."""
    protector = Protector(
        verifier,
        public_base_url=public_base_url,
        revocation=revocation,
        # The error never contains the token or claim values; never log the Authorization or DPoP header.
        on_error=lambda err: log.warning("rejected: %s", err),
    )
    app = FastAPI()
    install(app)  # problem+json responses and one WWW-Authenticate header per challenge

    @app.get("/api/public")
    def public() -> dict[str, Any]:
        # No token needed.
        return {"message": "Hello from a public endpoint."}

    @app.get("/api/private")
    def private(claims: AccessClaims = Depends(RequireAuth(protector))) -> dict[str, Any]:
        # Any valid access token for this API.
        return {"message": "Hello from a private endpoint.", "sub": claims.subject}

    @app.get("/api/messages")
    def messages(claims: AccessClaims = Depends(RequireAuth(protector, scopes=["messages:read"]))) -> dict[str, Any]:
        # A valid access token that also carries the messages:read scope.
        return {"messages": ["Hello from a scoped endpoint."]}

    # token.revoked deliveries from KeenAuth.
    if webhook_secrets:
        if not isinstance(revocation, MemoryRevocationCache):
            raise ValueError("the token.revoked webhook needs a MemoryRevocationCache to feed")
        app.include_router(webhook_router(RevocationWebhook(webhook_secrets, revocation)))
    else:
        log.warning("KEENAUTH_WEBHOOK_SECRET is not set: token.revoked events are not received, so revoked tokens stay valid until they expire")

    return app
```

`app_from_env` は設定から実際の verifier を構築します。verifier は、最初にトークンを見たときに環境から署名鍵を取得します。依存関係は同期的なので、FastAPI はそれをスレッドプールで実行します。

```python
def app_from_env() -> FastAPI:
    """The production app: KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required."""
    issuer = os.environ.get("KEENAUTH_ISSUER", "")
    audience = os.environ.get("KEENAUTH_AUDIENCE", "")
    # The URL clients call: DPoP proofs are checked against it, never against the Host header.
    public_base_url = os.environ.get("PUBLIC_BASE_URL", "")
    if not issuer or not audience or not public_base_url:
        raise SystemExit("KEENAUTH_ISSUER, KEENAUTH_AUDIENCE and PUBLIC_BASE_URL are required")
    # The verifier refuses an issuer under the reserved .invalid domain, so a test token can never pass here.
    verifier = Verifier(VerifierConfig(issuer=issuer, audience=audience))
    # Fed by the token.revoked webhook.
    revocation = MemoryRevocationCache()
    # The signing secret (idp_whsec_...) of the webhook endpoint that delivers token.revoked; comma separated while rotating.
    webhook_secrets = [x.strip() for x in os.environ.get("KEENAUTH_WEBHOOK_SECRET", "").split(",") if x.strip()]
    return create_app(verifier, public_base_url=public_base_url, revocation=revocation, webhook_secrets=webhook_secrets)
```

## 4. API を実行して呼び出す

```sh
export KEENAUTH_ISSUER=https://your-environment.eu.keenauth.app
export KEENAUTH_AUDIENCE=https://api.example.com
export PUBLIC_BASE_URL=https://api.example.com
uvicorn main:app_from_env --factory --port 8080
```

最初の呼び出しのために、`messages:read` スコープを API に対して持つマシン間アプリケーションの client credentials grant でトークンを取得します:

```sh
ACCESS_TOKEN=$(curl -s -u "$CLIENT_ID:$CLIENT_SECRET" \
  -d grant_type=client_credentials \
  -d resource=https://api.example.com \
  -d scope=messages:read \
  https://your-environment.eu.keenauth.app/oauth2/token | jq -r .access_token)

curl -i http://localhost:8080/api/messages -H "Authorization: Bearer $ACCESS_TOKEN"
```

## 5. ネットワークなしでテストする

`keenauth.testing` は公開テストキーで署名したトークンを発行し、`testing.verifier` はそのトークンだけを受け入れます。テストはその verifier を `create_app` に渡し、Starlette のテストクライアントでアプリを呼び出すため、ネットワークも KeenAuth アカウントも不要です。[オフラインテストキット](/docs/sdk/testing)を参照してください。

```python
@pytest.fixture()
def api() -> Api:
    cache = MemoryRevocationCache()
    app = create_app(
        testing.verifier(audience=AUDIENCE),
        public_base_url=BASE_URL,
        revocation=cache,
        webhook_secrets=[testing.webhook_secret()],
    )
    return TestClient(app, base_url=BASE_URL), cache
```
```python
def test_scope(api: Api) -> None:
    client, _ = api
    no_scope = testing.mint_access_token(AUDIENCE)
    assert_problem(client.get("/api/messages", headers=bearer(no_scope)), 403, "insufficient_scope", 'scope="messages:read"')
    with_scope = testing.mint_access_token(AUDIENCE, claims={"scope": "profile messages:read"})
    assert client.get("/api/messages", headers=bearer(with_scope)).status_code == 200
```

```sh
pip install --no-deps -r requirements-dev.txt
python -m pytest -q
```

この例のテストは、有効なトークンが受け入れられること、そして、トークンがない・期限切れ・audience 不一致、スコープ不足、URL クエリ内のトークン、取り消されたトークンが、以下のステータスで拒否されることを確認します。また、`PUBLIC_BASE_URL` 向けの proof を付けた DPoP バインドのトークンを送信し、`app_from_env` がテスト用 issuer を拒否することも確認します。テストでは署名された `token.revoked` 配信も送信します。有効な配信はトークンを取り消し、偽造・改ざん・古い配信は `400` を受け取って何も取り消さず、同じ配信を繰り返した場合は効果が 1 回だけになります。

## レスポンス

| リクエスト | ステータス | 本文の `code` |
|---|---|---|
| `Authorization` ヘッダがない | `401` | `missing_token` |
| 期限切れのトークン、別の API 向けのトークン、不正な署名、取り消されたトークン | `401` | `invalid_token` |
| 必要なスコープがないトークン | `403` | `insufficient_scope` |
| URL クエリ内のトークン（`?access_token=`） | `400` | `invalid_request` |

本文は `application/problem+json` で、`WWW-Authenticate` ヘッダには標準のチャレンジが入ります。完全な表は [ミドルウェアで API を保護する](/docs/sdk/protect-api#応答) にあります。

## プロキシの背後

クライアントが DPoP を使用する場合、ミドルウェアは各 proof がクライアントの呼び出した URL 向けに作られたことを確認します。`PUBLIC_BASE_URL` を、クライアントが呼び出す外部 URL に設定してください（例: `https://api.example.com`）。プロキシやロードバランサの背後にある場合も同様です。これは必須で、`https` を使用しなければなりません（`http` はローカルの `localhost`、`127.0.0.1`、`[::1]` の場合のみ使用できます）。設定がないとアプリは起動しません。ミドルウェア自体は `X-Forwarded-*` ヘッダを一切読みません。ただし Uvicorn は、接続元が `--forwarded-allow-ips` 内のアドレスである場合（既定では `127.0.0.1,::1`）、リクエスト URL に `X-Forwarded-Proto` を適用します。`PUBLIC_BASE_URL` を設定すると、チェックされる URL はそれに依存しなくなります。

## 取り消されたトークン

アクセストークンは期限切れまで有効なままなので、KeenAuth はトークンが取り消されたときに API へ通知します。この例では、インメモリの失効キャッシュをプロテクタに渡し、それにデータを供給する `token.revoked` webhook を `POST /webhooks/keenauth`（`webhook.py`）で提供します。

1. ダッシュボードで、URL `https://<your host>/webhooks/keenauth` を持ち `token.revoked` を購読する webhook エンドポイントを作成し、その署名シークレットを `KEENAUTH_WEBHOOK_SECRET` としてエクスポートします。
2. ハンドラは生のボディに対して SDK で署名を検証し、署名が無効な場合、またはタイムスタンプが 5 分以上ずれている場合は `400` を返します。イベントのセレクタ（`access_token`、`session`、`user`、`client`、`org`、`org_user`、`client_user`）を失効マーカーに対応付けて `204` を返します。その後ミドルウェアは、`not_before` より前に発行された一致するトークンをすべて拒否します。
3. KeenAuth は少なくとも 1 回配信するため、ハンドラは各 `webhook-id` を一度だけ適用します。マーカーを保存できない場合は `503` を返し、KeenAuth はイベントを再配信します。それ以外のイベントタイプや、どのアクセストークンにも一致しないセレクタ（`refresh_family`、`api`）は無視します。

ハンドラは、イベント ID、セレクタの種類、理由をログに記録しますが、ボディ、署名、シークレットは記録しません。各ワーカープロセスは独自のキャッシュを持ちます。複数のワーカーやインスタンスを使う場合は、代わりに共有ストアを使ってください。[失効キャッシュ](/docs/sdk/protect-api#取り消しキャッシュtokenrevoked)を参照してください。

## 次の手順

- [ミドルウェアで API を保護する](/docs/sdk/protect-api): Flask、ステップアップ要件、DPoP、およびすべてのオプション。
- [トークンを検証する](/docs/sdk/token-verification): ミドルウェアの下で動作する verifier。
- [オフラインテストキット](/docs/sdk/testing)。