TOTPを実装する
0. はじめに
この記事は、学んだ技術や知識についてを備忘録としてまとめることを目的にしている。今回はTOTPについてまとめる。
> 注意
> 本記事のコードは、仕組みを確かめるための検証用です。パスワードの扱いなど、実際のシステムに組み込むときに必要な処理を省いています。実際のシステムで使う場合は、4.2 を読んでください。
1. 開発環境
<!-- TODO: Mac で確認して差し替える -->
- macOS
- uv 0.7.8
- Python 3.13.5(uv で用意する)
- FastAPI 0.162.0、uvicorn 0.38.0
- PyJWT 2.10.1
- pyotp 2.10.0
- qrcode 8.2
- cryptography 46.0.3
- SQLite(Python 標準の sqlite3 と、macOS に入っている sqlite3 コマンド)
- スマートフォンの認証アプリ(Google Authenticator、Microsoft Authenticator など)
2. 基本概念
2.1 多要素認証
本人を確かめる手がかり(認証の要素)は、次の3種類に分けられる。
| 要素 | 何で確かめるか | 例 |
|---|---|---|
| 知識要素 | 本人だけが知っていること | パスワード、暗証番号 |
| 所持要素 | 本人だけが持っているもの | スマートフォン、IC カード |
| 生体要素 | 本人の体の特徴 | 指紋、顔 |
種類の違う要素を2つ以上組み合わせる認証を、多要素認証(MFA)という。
ID・パスワードだけのログインは知識要素1つに頼っている。パスワードは漏えいや使い回し、推測によって他人に知られると、それだけでログインされてしまう。そこで知識要素に所持要素や生体要素を足し、パスワードを知られただけではログインできないようにする。
所持要素を足す場合は、利用者が端末を持っていることをネットワーク越しにサーバーへ示す必要がある。その方法の1つが端末に使い捨てのコードを作らせるワンタイムパスワードである。
2.2 ワンタイムパスワード(OTP)
2.2.1 使い捨てのコード
所持要素を確かめるには、その端末にしか作れない値を入力してもらえばよい。ただしその値が毎回同じだと、パスワードと同じく一度知られただけで使われ続けてしまう。そこで値をログインのたびに変え、一度使った値は使えないようにする。これがワンタイムパスワード(OTP)である。
OTP(One-Time Password) は、次の3つの手順で「端末を持っていること」を確かめる。
1. 端末とサーバーが同じ秘密の値(秘密鍵)をあらかじめ共有しておく
2. 端末は秘密鍵と「その時だけの値」からコードを計算して表示し、利用者がそれを入力する
3. サーバーも同じ秘密鍵と同じ「その時だけの値」から同じ計算をし、入力されたコードと比べる
秘密鍵を持たない端末は正しいコードを作れない。このため、コードが一致すれば利用者は秘密鍵の入った端末を手元に持っていると判断できる。またコードはその時だけのものなので、盗み見られてもあとで使い回すことはできない(有効な間に使い回されるのを防ぐ方法は2.4.3 で説明する)。
「その時だけの値」に何を使うかで方式が分かれる。
回数(カウンター)を使うのが HOTP(RFC 4226)、
時刻を使うのが TOTP(RFC 6238)である。
2.2.2 HOTP:秘密鍵とカウンターから作る
TOTP は HOTP を元にしている。HOTP(HMAC-based One-Time Password)とは、秘密鍵 K と、サーバーと端末がそれぞれ数えるカウンター C からコードを作る方式(RFC 4226 §5.2)。
HOTP(K, C) = Truncate(HMAC-SHA-1(K, C))
HOTP のコードは、次の手順で作る。
1. HMAC-SHA-1(K, C):秘密鍵 K とカウンター C から、20バイトの値を作る
2. Truncate:20バイトのうち、最後の1バイトの下位4ビットが示す位置から4バイトを取り出し、先頭の1ビットを落として31ビットの数にする(RFC 4226 §5.3)
3. その数を $10^6$ で割った余りが、6桁のコードになる。桁数は最低6桁と決められている(RFC 4226 §5.3)
sequenceDiagram
participant A as 認証アプリ(端末)
participant U as 利用者
participant S as サーバー
Note over A,S: 両方が同じ秘密鍵 K と、同じ値のカウンター C を持っている
A->>A: 端末のカウンター C で HOTP(K, C) を計算
A->>A: 端末のカウンター C を1つ進める
A->>U: 6桁のコードを表示
U->>S: コードを入力して送る
S->>S: サーバーのカウンター C で HOTP(K, C) を計算
S->>S: 入力されたコードと比べる
S->>S: 一致したら、サーバーのカウンター C を1つ進める
HOTP では、端末はコードを作るたびにカウンターを1つ進め、サーバーはコードの照合に成功するたびにカウンターを1つ進める。このため、端末でコードを作っただけで使わなかった場合、端末のカウンターだけが進み、サーバーと端末のカウンターがずれる。ずれると、端末が表示したコードとサーバーが計算したコードが一致せず、認証できなくなる。
RFC 4226 は、サーバーが先の数個のカウンターまで試し、一致したらカウンターを合わせ直す方法を定めている(RFC 4226 §7.4)。ただし、試す範囲を超えてずれると、合わせ直せない。
そこでカウンターの代わりにサーバーと端末が同じ値を自然に持てる時刻を用いた方式が考案された。それが TOTP である。
<details><summary>用語解説:HMACとは</summary>
HMAC(Hash-based Message Authentication Code)は、秘密鍵とデータからハッシュ関数を使って決まった長さの値を作る計算である(RFC 2104 §2)。次の2つの性質がある。
- 同じ鍵と同じデータからは必ず同じ値になる
- 鍵を知らなければその値を計算できない
HMAC-SHA-1 はハッシュ関数に SHA-1 を使う HMAC で、結果は20バイトになる。HOTP はこの性質を使って、秘密鍵を持つ端末とサーバーだけが同じコードを作れるようにしている。
</details>
2.2.3 TOTP:カウンターを時刻に置き換える
TOTP(Time-based One-Time Password)は、HOTP のカウンター C を時刻から求めた番号 T に置き換えたもの(RFC 6238 §4.2)。
T = floor(Unix時刻 / 30)
TOTP(K) = HOTP(K, T)
T は時刻を30秒ごとに区切った通し番号(タイムステップ)である。同じ30秒の間は T が同じなので、同じコードになる。コマンドを実行した時点(現在時刻)の T は、次のコマンドで確かめられる。
python3 -c "import time; print(int(time.time()) // 30)"
サーバーと端末はそれぞれ自分の時計から T を求め、それぞれでコードを計算する。
sequenceDiagram
participant A as 認証アプリ(端末)
participant U as 利用者
participant S as サーバー
Note over A,S: 両方が同じ秘密鍵 K を持っている
A->>A: 端末の現在時刻で TOTP(K) を計算
A->>U: 6桁のコードを表示
U->>S: コードを入力して送る
S->>S: サーバーの現在時刻で TOTP(K) を計算
S->>S: 入力されたコードと比べる
端末とサーバーの間で通信するのは利用者が入力したコードだけで、端末はネットワークにつながっていなくても、コードを作れる。
TOTP の既定値
TOTP では、コードの有効時間・桁数・HMAC の種類を選べる。既定値は次のとおりである。
| 値 | 既定値 | 意味 | 根拠 |
|---|---|---|---|
| 有効時間 | 30秒 | 1つのコードが有効な時間の単位。短いほど盗まれたコードの寿命は短くなるが、入力が間に合わなくなる。その釣り合いとして推奨されている | RFC 6238 §5.2 |
| 桁数 | 6桁 | 勘で当たる確率は、1回あたり 1/1,000,000 | RFC 4226 §5.3 |
| HMAC の種類 | HMAC-SHA-1 | HOTP と同じ。TOTP では HMAC-SHA-256 や HMAC-SHA-512 に変えてもよい | RFC 6238 §1.2 |
本記事では3つとも既定値を使う。既定値から変えない理由は、2.3 の補足で説明する。
2.3 秘密鍵の受け渡し(登録)
TOTP ではサーバーと端末が同じ秘密鍵を持っていることが前提になる。では、秘密鍵をどうやって端末に渡すか。
2.3.1 登録は最初の1回だけ
秘密鍵を端末に渡す作業を、登録という。利用者ごとに最初の1回だけ行う。
1. サーバーが、利用者ごとの秘密鍵を作り、保存する
2. 秘密鍵を QR コードにして画面に表示する
3. 利用者が、認証アプリで QR コードを読み取る
4. 認証アプリに表示されたコードを入力してもらい、正しければ登録を確定する
秘密鍵が通信路を通るのは、この1回だけである。以降のログインでは6桁のコードしか送らない。
4 で確認するのは、読み取りに失敗していた場合や別の端末で読み取ってしまった場合に、確定する前に気づけるようにするためである。確認せずに確定すると本人がログインできなくなる。
秘密鍵の長さは最低128ビット、推奨160ビットとされている(RFC 4226 §4)。
2.3.2 otpauth:// URI
QR コードの中身は、次の形式の文字列(URI)である。
otpauth://totp/{issuer}:{account}?secret={秘密鍵}&issuer={issuer}
| 項目 | 説明 |
|---|---|
| totp | 種類。時刻ベースなら totp、カウンターベースなら hotp |
| {issuer}:{account} | 認証アプリに表示される名前。サービス名とアカウント名 |
| secret | 秘密鍵。Base32 で書く |
| issuer | サービス名。ラベルの issuer と同じにする |
| algorithm・digits・period | 省略時は SHA1・6桁・30秒 |
この形式は RFC ではなく、Google Authenticator が公開した仕様(Key Uri Format)である。多くの認証アプリが対応しており、事実上の標準になっている。
QR コードは、この URI をそのまま画像にしたものである。つまり、QR コードの画像は秘密鍵そのものである。よってスクリーンショットを残したり、外部の QR 生成サービスに URI を送ったりしてはいけない。
<details><summary>補足:なぜ既定値から変えないのか</summary>
Key Uri Format の仕様ページには、Google Authenticator は algorithm と period を無視する、と書かれている(最終更新は2018年で、現行版での扱いは未確認)。サーバーだけ SHA-256 や60秒にしても、アプリが既定値で計算すればコードは一致しない。このため SHA-1・6桁・30秒の既定値から変えないのが安全である。
</details>
これでサーバーと端末が同じ秘密鍵を持ち、同じ計算ができるようになった。ではサーバーはコードを照合するとき、何に気をつけるか。
2.4 サーバー側の照合
2.4.1 時刻のずれを許す
サーバーと端末の時刻が少しでもずれているとT が1つ違い、コードが一致しない。また30秒の切り替わりの直前にコードを入力すると、届くまでに次の T に変わってしまう。
そこでサーバーは照合する時点(現在時刻)の T のコードだけでなく、その前後のステップのコードも受け付ける。RFC 6238 は通信の遅れとして多くても1ステップの許容を推奨している(RFC 6238 §5.2)。前後1ステップを許すと「1つ前」「現在」「1つ後」の3つのステップのコードを受け付けることになる。
ただし受け付けるコードが増えると、次の2つの副作用が出る。
| 副作用 | 内容 | 対策 |
|---|---|---|
| 当たりやすくなる | 1回あたりの当たる確率が、1/1,000,000 から 3/1,000,000 になる | 2.4.2 |
| 使い回しやすくなる | 1つのコードが、最大で約90秒間有効になる | 2.4.3 |
2.4.2 試行回数を制限する
6桁は100万通りしかないので、何回でも試せるなら総当たりでいずれ当たる。そこで失敗の回数に上限を設けて上限に達したら一定時間ロックする(RFC 4226 §7.3)。
- 回数はログインのたびではなく利用者ごとに数える。ログインし直すだけで回数が戻ると制限の意味がない
- 上限は使い勝手を損なわない範囲でなるべく小さくする。NIST SP 800-63B は、連続失敗の上限を100回までとしている(NIST SP 800-63B-4 §3.2.2)
2.4.3 同じコードを2回使えなくする
コードが最大で約90秒有効だと利用者が入力したコードを肩越しに見てすぐに使われるおそれがある(リプレイ攻撃)。RFC 6238 は一度成功したコードの2回目を受け付けてはならない(MUST NOT)としている(RFC 6238 §5.2)。
これを実装するには、利用者ごとに「最後に成功したタイムステップ番号」を保存し、それ以下のステップのコードを拒否する。
2.4.4 秘密鍵を暗号化して保存する
サーバーは照合のたびに秘密鍵からコードを計算する。このため元の秘密鍵を持っておく必要がある。
| | パスワード | TOTP の秘密鍵 |
|---|---|---|
| 照合の方法 | 入力をハッシュ化して、保存した値と比べる | 秘密鍵からコードを計算して、入力と比べる |
| 保存の方法 | ハッシュ化(元に戻せない) | 暗号化(照合のときに復号する) |
| 漏れたとき | ハッシュから元に戻すのは難しい | 秘密鍵があれば、いつでも正しいコードを作れる |
秘密鍵はハッシュ化できないので暗号化して保存し、照合するときだけ復号する(RFC 6238 §5.1)。暗号化の鍵は、DB と別の場所に置く。
本記事では、暗号化に Python の cryptography ライブラリに入っている Fernet を使う。
<details><summary>用語解説:Fernetとは</summary>
Fernet は、Python の cryptography ライブラリに入っている共通鍵暗号の仕組みである(cryptography Fernet)。1つの鍵で、暗号化と復号の両方を行う。
- 暗号化には AES(128ビットの鍵、CBC モード)を使う
- 改ざんされていないかを確かめるために、HMAC-SHA256 を付ける
- 鍵は、32バイトの値を Base64 で書いた44文字の文字列である
- 暗号文は gAAAAA で始まる
AES と HMAC を自分で組み合わせると、使い方を間違えやすい。Fernet は、それらを正しく組み合わせた決まった手順として使えるので、暗号の細かい設定を考えずに済む。
</details>
2.5 既存のログインに組み込む
TOTP を足すと、ログインでは「パスワード」と「コード」の2つを確かめることになる。2つを1回のリクエストで送らせる方法と、パスワードを確かめてからコードを求める2段階の方法がある。2段階にすると、パスワードが違う人にはコードの入力欄を見せずに済み、TOTP を登録していない利用者は1段目でログインを終えられる。
TOTP を足したあと、利用者がパスワードやコードを送る場面は、次の3つになる。
| 場面 | 利用者が送るもの | TOTP か | 行う回数 | 何を確かめるか | 通ったら |
|---|---|---|---|---|---|
| ログインの1段目 | ID・パスワード | TOTP ではない | ログインのたび | パスワードを知っているか(知識要素) | TOTP を登録していればコードを求める。登録していなければログイン完了 |
| 登録の確定(2.3.1 の手順4) | 認証アプリの6桁のコード | TOTP | 利用者ごとに最初の1回だけ | QR を正しく読み取れたか。つまり、サーバーと認証アプリが同じ秘密鍵を持てたか | TOTP を有効にする。以後のログインで2段目を求めるようになる |
| ログインの2段目 | 認証アプリの6桁のコード | TOTP | ログインのたび | 秘密鍵の入った端末を持っているか(所持要素) | ログイン完了 |
ログインの「2段階」は、1段目(パスワード)と2段目(TOTP)の組み合わせである。種類の違う2つの要素を1段ずつ確かめることで、多要素認証になる。
認証アプリのコードを入力する場面は、登録の確定とログインの2段目の2つである。どちらも同じ TOTP のコードで、照合のしかた(2.4)も同じである。違うのは、何のために確かめるかと、通ったあとにサーバーが行うことである。登録の確定はログインではないので、ここでコードを入力しても、ログインしたことにはならない。
2段階にする場合、サーバーは「この利用者はパスワードが通った」ことを、2段目のリクエストまで覚えておく必要がある。そこで、1段目でパスワードが通ったら、短い期限(本記事では5分)の証明を発行する。これをTOTP待ちトークンと呼ぶ。2段目でコードも通ったら、ログイン済みであることを示す証明を発行する。これをアクセストークンと呼ぶ。TOTP待ちトークンは、アクセストークンの代わりに使えてはいけない。使えると、コードを入れずにログインできてしまう。
2.6 TOTPで防げること・防げないこと
| 攻撃・事故 | TOTP で防げるか | 理由・対策 |
|---|---|---|
| パスワードの漏えい | 防げる | パスワードだけではログインできない |
| コードの盗み見 | 防げる | 使用済みのコードは使えない(2.4.3) |
| 総当たり | 防げる | 試行回数の制限(2.4.2) |
| フィッシング | 防げない | 偽のログイン画面に入れたパスワードとコードを、攻撃者がその場で本物の画面に入れればログインできる |
| サーバーからの秘密鍵の漏えい | 防げない | 秘密鍵があれば、誰でも正しいコードを作れる。暗号化して保存する(2.4.4) |
| 認証アプリのクラウド同期 | 弱まる | 秘密鍵が別の端末にも入り、「その端末を持っている」ことの証明にならなくなる |
| 端末の紛失・盗難 | 一部防げる | パスワードも要るので、端末だけではログインできない |
NIST SP 800-63B は、OTP にはフィッシング耐性が無いと明記している(NIST SP 800-63B-4 §3.2.5)。フィッシングまで防ぐには FIDO2 / WebAuthn(パスキー)のように、アクセス先のサイトを確かめてから認証する方式が要る。
<details><summary>補足:パスワードのみ・TOTP・FIDO2 / WebAuthn の比較</summary>
| | パスワードのみ | パスワード+TOTP | FIDO2 / WebAuthn |
|---|---|---|---|
| 確かめる要素 | 知識 | 知識+所持 | 所持(+生体・暗証番号) |
| サーバーに送るもの | パスワード | パスワードとコード | 端末の秘密鍵で作った署名(秘密鍵は送らない) |
| サーバーが持つ秘密 | パスワードのハッシュ | パスワードのハッシュと、TOTP の秘密鍵 | 公開鍵だけ |
| フィッシング耐性 | 無い | 無い | ある |
| 導入のしやすさ | ― | 既存のログインに足しやすい。認証アプリは無料 | 端末やブラウザの対応が要る |
</details>
TOTP は既存の ID・パスワードのログインに手軽に所持要素を足せる方式である。3章以降では本章で説明した登録(2.3)と照合(2.4)を実装し、2.5 の2段階の形で既存のログインに組み込む。
2.7 その他の用語
| 用語 | 意味 |
|---|---|
| Unix 時刻 | 1970-01-01 00:00:00 UTC からの経過秒数。タイムゾーンに左右されないので、日本の端末と海外のサーバーでも同じ T になる |
| HMAC | 秘密鍵を使って、データを決まった長さの値に変換する計算。鍵を知らないと、同じ結果を作れない |
| Base32 | データを英大文字 A〜Z と数字 2〜7 の32種類の文字だけで表す方法(RFC 4648 §6)。紛らわしい 0 と 1 を使わないので、手入力しやすい |
| ハッシュ化 | データを、元に戻せない形に変換すること |
| Fernet | Python の cryptography に入っている共通鍵暗号の仕組み。1つの鍵で暗号化と復号を行い、改ざんの検知も付いている |
| JWT | 署名付きのトークン。ログインに成功したときにサーバーが発行し、以降のリクエストに付けてログイン済みであることを示す |
| NTP | ネットワーク経由で、機器の時計を正しい時刻に合わせる仕組み。TOTP は時刻が命なので、サーバーの時計は NTP で合わせておく |
3. TOTP実装
本章では、2章で説明した TOTP の仕組みを使って、ID・パスワードのログインに TOTP の確認を足す小さな API を作る。次の順に進める。
| 節 | 内容 |
|---|---|
| 3.1 | 作るものの全体像(ファイル構成、ログインの流れ、エンドポイント、テーブル、設計値) |
| 3.2 | 事前準備(Python の仮想環境、パッケージ、暗号化の鍵) |
| 3.3 | TOTP の登録と照合の処理を書く(totp.py) |
| 3.4 | totp.py を使って API を書く(main.py) |
3.1 作るものの全体像
この記事で作るものは、2.5 で説明した2段階のログインである(流れは 3.1.2 の図を参照)。
TOTP を登録していない利用者には、これまでどおりアクセストークンを返す。TOTP の登録は、ログインした本人が行う(3.1.3 の /totp/enroll)ため、最初から全員に TOTP を求めると、まだ登録していない利用者はログインできず、登録もできなくなるからである。また既存のログインに足す場合、全員が一度に登録を終えられるわけではない。
ただし、TOTP を登録していない利用者はパスワードだけでログインできるままである。実際のシステムに組み込む場合は、全員に登録を必須にするか、いつまでに登録してもらうかを決める必要がある(4.2.1)。
3.1.1 ファイル構成
~/totp-demo
├── requirements.txt # 使うパッケージ(3.2)
├── totp.py # TOTP の登録と照合(3.3)
├── main.py # API(3.4)
├── demo.db # 起動すると作られる SQLite の DB
└── qr.svg # 4.1 で保存する QR コード
TOTP そのものの処理は totp.py に、API の処理は main.py に分けた。
3.1.2 ログインの流れ
sequenceDiagram
autonumber
participant U as 利用者
participant S as サーバー
U->>S: POST /login(ID・パスワード)
S->>S: パスワードを確かめる
alt TOTP を登録していない利用者
S->>U: アクセストークン(ログイン完了)
else TOTP を登録済みの利用者
S->>U: TOTP待ちトークン(5分)
U->>S: POST /login/totp(TOTP待ちトークン+6桁のコード)
S->>S: コードを照合する
S->>U: 正しければアクセストークン(ログイン完了)
end
TOTP待ちトークンは、「パスワードは通ったが、TOTP はまだ」という状態を表す。アクセストークンとして使えてはいけない(3.4.1)。
3.1.3 エンドポイント一覧
POST /login と POST /login/totp は3.1.2 の図の工程に対応する。残りの3つはログインしたあと(アクセストークンを持っている状態)で使う。
| エンドポイント | 必要なもの | 3.1.2 の図の工程 | 役割 |
|---|---|---|---|
| POST /login | ID・パスワード | 1〜4 | パスワードを確かめ(2)、<br>TOTP を登録していなければアクセストークン(3)、<br>登録済みなら TOTP待ちトークン(4)を返す |
| POST /login/totp | TOTP待ちトークン、コード | 5〜7 | コードを照合し(6)、<br>正しければアクセストークンを返す(7) |
| GET /me | アクセストークン | ―(ログイン後) | ログイン中の利用者を返す |
| POST /totp/enroll | アクセストークン | ―(ログイン後) | 秘密鍵を作り、QR コードを返す |
| POST /totp/activate | アクセストークン、コード | ―(ログイン後) | 認証アプリのコードが正しければ、TOTP を有効にする |
3.1.4 usersテーブル
ID・パスワードの列にTOTP 用の5列を足す。
| 列 | 内容 |
|---|---|
| totp_secret | 秘密鍵。暗号化して保存する |
| totp_enabled | 1 なら、ログインで TOTP を求める |
| totp_last_used_step | 最後に成功したタイムステップ番号。再利用の防止に使う |
| totp_failures | 続けて失敗した回数 |
| totp_locked_until | ロックが解ける時刻(Unix 時刻) |
3.1.5 採用する設計値
| 項目 | 値 | なぜ必要か | 根拠 |
|---|---|---|---|
| 時刻のずれ | 前後1ステップ(±30秒)まで許す | 端末の時計が少しずれていても、ログインできるようにする | RFC 6238 §5.2 |
| 再利用の防止 | 最後に成功したステップ以下のコードは拒否する | 盗み見たコードを、有効なうちに使い回されないようにする | RFC 6238 §5.2 |
| 試行回数の制限 | 5回続けて失敗したら15分ロックする | 100万通りを総当たりされないようにする | RFC 4226 §7.3 |
| 秘密鍵の保存 | 暗号化して DB に保存する。暗号化の鍵は環境変数で渡す | DB から漏れても、すぐには使えないようにする | RFC 6238 §5.1 |
| 方式 | SHA1・6桁・30秒 | 認証アプリとの互換性 | RFC 6238 の既定値 |
| 秘密鍵の長さ | 160ビット | 推測されないようにする | RFC 4226 §4 |
<details><summary>補足:「5回で15分」の根拠</summary>
「5回で15分」は業界の標準値ではなく、本記事で採用した値である。NIST SP 800-63B では連続失敗の上限を100回までとしている(§3.2.2)。この設定だと1日に試せるのは最大480回で前後1ステップを許しても、1日で当たる確率は約0.14%(480 × 3 / 1,000,000)。
</details>
3.2 事前準備(環境構築)
3.2.0 作業ディレクトリとコードの扱い
作業ディレクトリは ~/totp-demo とする。以降のコマンドは、すべてこのディレクトリの中で実行する。
mkdir -p ~/totp-demo
cd ~/totp-demo
3.3 と 3.4 のコードはファイルごとに数回に分けて載せている。「totp.py(1/4)」のように番号を付けているので、上から順に1つのファイルにつなげる。つなげた全体は各節の最後に折りたたんで載せている。
3.2.1 uvでPythonの仮想環境を作る
FastAPI 0.162 は Python 3.10 以上が必要である。macOS に最初から入っている python3 は 3.9 なので、そのままでは動かない。そこでuv で Python 3.13 の仮想環境を作る。
uv が入っているかを確かめる。入っていなければ、[uv の記事](TODO: リンク)を参照して入れる。
uv --version
uv 0.7.8 (0ddcc1905 2025-05-23)
仮想環境を作る。
uv venv --python 3.13
Using CPython 3.13.5 interpreter at: /usr/local/opt/python@3.13/bin/python3.13
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate
筆者の Mac には Homebrew で入れた Python 3.13 があったので、uv はそれを使った。Python 3.13 が入っていない場合は、uv が自動でダウンロードしてから仮想環境を作る。
.venv ができる。この記事では activate せず、.venv/bin/python のようにパスを書いて使う。どの Python で動かしているかがコマンドを見れば分かるようにするためである。
3.2.2 パッケージを入れる
このデモで使うパッケージを、requirements.txt というファイルに書き出しておく。requirements.txt は、必要なパッケージとそのバージョンを1行に1つずつ並べたテキストファイルである。こうしておくと次の2つの利点がある。
- 1つのコマンドで、必要なパッケージをまとめて入れられる
- == でバージョンを決めておくので、読者の環境にも記事と同じバージョンが入る。バージョンが違うと、動きや実行結果が記事と変わることがある
~/totp-demo に、次の内容で requirements.txt を作る。
fastapi==0.162.0
uvicorn==0.38.0
pyjwt==2.10.1
pyotp==2.10.0
qrcode==8.2
cryptography==46.0.3
| パッケージ | 用途 |
|---|---|
| fastapi、uvicorn | API を作って動かす |
| pyjwt | ログインのトークン(JWT)を作る |
| pyotp | TOTP の秘密鍵の作成と、コードの計算 |
| qrcode | otpauth:// URI を QR コードにする |
| cryptography | 秘密鍵を暗号化する(Fernet) |
以下のコマンドで入れる。
uv pip install -r requirements.txt
Resolved 20 packages in 898ms
Installed 20 packages in 20ms
+ cryptography==46.0.3
+ fastapi==0.162.0
+ pyjwt==2.10.1
+ pyotp==2.10.0
+ qrcode==8.2
+ uvicorn==0.38.0
(ほかの依存パッケージは省略)
3.2.3 暗号化の鍵を環境変数に入れる
秘密鍵を DB に保存するときに使う暗号化の鍵を作る。
環境変数は、それを設定したターミナルの中でだけ有効である。このため、次のコマンドは、4.1 でサーバーを起動するときにも使い続けるターミナルで実行する(4.1 では、このターミナルを「ターミナルA」と呼ぶ)。
export TOTP_ENCRYPTION_KEY=$(.venv/bin/python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())")
入ったかを確かめる。値は実行のたびに変わる。
echo $TOTP_ENCRYPTION_KEY
44文字の英数字(末尾は =)が表示されればよい。
<details><summary>補足:なぜ環境変数で渡すのか</summary>
暗号化の鍵がソースや DB と同じ場所にあると、DB と一緒に漏れたときに、暗号化した意味がなくなる。このため、鍵はソースにも DB にも置かず、起動するときに外から渡す。
</details>
<details><summary>補足:ターミナルを閉じたとき</summary>
環境変数は、ターミナルを閉じると消える。新しい鍵を作り直すと、前の鍵で暗号化した秘密鍵は復号できない。その場合は、demo.db を消して最初からやり直す。認証アプリに残った登録も消しておく。
</details>
3.3 TOTPの処理を書く(totp.py)
本節では、2章で説明した登録(2.3)と照合(2.4)の処理を、totp.py に書く。totp.py は、3.4 の API から呼ばれる部品であり、次の順に組み立てる。
| 節 | 書くもの | 対応する2章の説明 |
|---|---|---|
| 3.3.1 | 設計値の定数と、暗号化の準備 | 2.4.4(秘密鍵の暗号化) |
| 3.3.2 | 登録:enroll() | 2.3(秘密鍵の受け渡し) |
| 3.3.3 | 一致したステップを求める:find_matched_step() | 2.4.1(時刻のずれを許す) |
| 3.3.4 | 照合:verify() | 2.4.2(試行回数の制限)、2.4.3(再利用の防止) |
~/totp-demo に totp.py を作り、3.3.1〜3.3.4 のコードを順に書いていく。
3.3.1 定数と暗号化の準備
totp.py の先頭では、このあとの関数が共通で使う値と道具を用意する。3.1.5 で決めた設計値(許容するステップ数、ロックまでの回数と時間)を定数として置き、秘密鍵の暗号化と復号に使う Fernet を、3.2.3 で環境変数に入れた鍵で準備する。
"""TOTP の登録と照合を行う部品。
main.py(API)から呼ばれる。このファイルには、次の3つの関数がある。
- enroll() : 登録。秘密鍵を作って DB に保存し、認証アプリに読ませる URI を返す
- find_matched_step() : 入力されたコードが、どのタイムステップのコードと一致するかを調べる
- verify() : 照合。コードが正しいかを確かめ、再利用の防止と試行回数の制限も行う
"""
# 環境変数を読むために使う(Python 標準)
import os
# 現在時刻(Unix 時刻)を得るために使う(Python 標準)
import time
# TOTP の秘密鍵の作成と、コードの計算を行うライブラリ
import pyotp
# 秘密鍵の暗号化と復号に使う仕組み(記事の 2.4.4)
from cryptography.fernet import Fernet
# ---------------------------------------------------------------------------
# 設計値(記事の 3.1.5 で決めた値)
# ---------------------------------------------------------------------------
# 認証アプリに表示されるサービス名。アプリには「TOTP-Demo(alice@example.com)」のように出る
ISSUER = "TOTP-Demo"
# 時刻のずれを、前後何ステップまで許すか。
# 1 なら「1つ前」「現在」「1つ後」の3つのステップのコードを受け付ける(1ステップ=30秒)
VALID_WINDOW = 1
# 何回続けて失敗したらロックするか
MAX_FAILURES = 5
# ロックする時間(秒)。15分 × 60秒 = 900秒
LOCK_SECONDS = 15 * 60
# ---------------------------------------------------------------------------
# 暗号化の準備
# ---------------------------------------------------------------------------
# 秘密鍵を DB に保存するときに使う、暗号化の鍵を読み込む。
# 鍵はソースコードにも DB にも書かず、起動するときに環境変数 TOTP_ENCRYPTION_KEY で渡す。
# 環境変数が無いと、ここで KeyError になり、サーバーは起動しない。
fernet = Fernet(os.environ["TOTP_ENCRYPTION_KEY"])
- 3.1.5 の設計値を定数にしている
- fernet は、2.4.4 で説明した Fernet で、秘密鍵の暗号化(3.3.2)と復号(3.3.4)に使う
- 鍵は 3.2.3 の環境変数から読む。環境変数が無いと KeyError になって起動しない
3.3.2 登録:秘密鍵を作り、暗号化して保存する
2.3.1 の登録の手順のうち、サーバー側の作業である「秘密鍵を作って保存して認証アプリに渡す otpauth:// URI を作る」部分を enroll() 関数として定義する。3.4.5 の /totp/enroll から呼ばれる。
def enroll(db, user):
"""登録:秘密鍵を作り、暗号化して DB に保存する。
戻り値は otpauth:// で始まる URI(文字列)。
main.py がこれを QR コードにして返し、利用者が認証アプリで読み取る。
この関数では、まだ TOTP を有効にしない(有効にするのは /totp/activate)。
"""
# 利用者ごとの秘密鍵を、ランダムに作る。
# 160ビットの値を Base32(英大文字 A〜Z と数字 2〜7)の32文字で表したもの
secret = pyotp.random_base32()
# 秘密鍵を暗号化して、DB に保存する。
# - fernet.encrypt() は bytes を受け取って bytes を返すので、
# .encode() で文字列 → bytes に、.decode() で bytes → 文字列に変換している
# - totp_last_used_step(最後に使ったステップ)を空に戻すのは、
# 秘密鍵を作り直したときに、前の秘密鍵の記録が残らないようにするため
# - SQL の ? には、後ろのタプルの値が順に入る(SQL インジェクションを防ぐ書き方)
db.execute(
"UPDATE users SET totp_secret = ?, totp_last_used_step = NULL WHERE id = ?",
(fernet.encrypt(secret.encode()).decode(), user["id"]),
)
# 認証アプリに渡す otpauth:// URI を作って返す。
# name はアカウント名(ここではメールアドレス)、issuer_name はサービス名。
# 暗号化する前の秘密鍵が URI に入るので、この URI(と QR)は人に見せてはいけない
return pyotp.TOTP(secret).provisioning_uri(name=user["email"], issuer_name=ISSUER)
- pyotp.random_base32() は160ビットの秘密鍵を Base32 の32文字で返す
- 保存する前に fernet.encrypt() で暗号化する。DB には暗号文だけが入る
- totp_last_used_step を空に戻すのは、秘密鍵を作り直したときに、前の秘密鍵で使ったステップが残らないようにするためである
- provisioning_uri() は、2.3.2 の otpauth:// URI を作る。name がアカウント名、issuer_name がサービス名になる
この関数では TOTP をまだ有効にしない。有効にするのは認証アプリに出たコードが通ってからである(3.4.5)。
3.3.3 一致したステップ番号を求める
照合の準備として入力されたコードがどのタイムステップのコードと一致するかを調べる関数 find_matched_step() を作る。2.4.1 で説明したとおり現在時刻のステップだけでなく、前後1ステップのコードも受け付ける。一致したステップの番号まで求めるのは、3.3.4 で再利用を防ぐのに使うためである。
def find_matched_step(secret, code, now):
"""入力されたコードが、どのタイムステップのコードと一致するかを調べる。
引数
secret : 秘密鍵(復号した Base32 の文字列)
code : 利用者が入力した6桁のコード(文字列)
now : 現在時刻(Unix 時刻。秒)
戻り値
一致したタイムステップ番号(整数)。どれとも一致しなければ None。
pyotp には verify() という照合用の関数もあるが、True / False しか返さず、
どのステップで一致したかが分からない。再利用を防ぐにはステップ番号が要るので、自分で調べる。
"""
# この秘密鍵でコードを計算する道具を作る(既定値は SHA-1・6桁・30秒)
totp = pyotp.TOTP(secret)
# 現在のタイムステップ番号を求める。
# T = floor(Unix 時刻 / 30)。// は「割り算の商の整数部分」を求める演算子
current_step = int(now) // totp.interval
# offset を -1, 0, +1 と変えながら、「1つ前」「現在」「1つ後」のステップのコードと比べる
for offset in range(-VALID_WINDOW, VALID_WINDOW + 1):
# totp.at(now, offset) は、現在から offset ステップずらした時刻のコードを返す。
# 比べるときは == ではなく strings_equal を使う。
# == は先頭から比べて、違う文字が見つかった時点で止まるので、
# かかった時間の差から「何文字目まで合っていたか」を推測される余地がある。
# strings_equal は、一致の度合いにかかわらず同じ時間で比べる
if pyotp.utils.strings_equal(code, totp.at(now, offset)):
return current_step + offset
# どのステップのコードとも一致しなかった
return None
前後1ステップを許すので「1つ前」「現在」「1つ後」の3つのステップのコードと比べ、一致したもののタイムステップ番号を返す。
<details><summary>補足:なぜ pyotp の verify() を使わないのか</summary>
pyotp には照合のための verify() がある。
pyotp.TOTP(secret).verify(code, valid_window=1)
しかしverify() は True / False しか返さず、どのステップで一致したかが分からない。再利用を防ぐ(3.3.4.2)には、一致したステップ番号を保存する必要がある。このためat() で各ステップのコードを作り、自分で比べている。
</details>
<details><summary>補足:比べるときに strings_equal を使う理由</summary>
== で比べると、先頭から何文字目まで一致したかによって比べる時間がわずかに変わる。この時間の差からコードを推測される余地がある。pyotp.utils.strings_equal() は一致の度合いにかかわらず同じ時間で比べる。
</details>
3.3.4 照合:再利用の防止と試行回数の制限
照合の本体である verify() を作る。3.3.3 で一致したステップを求めたうえで、2.4.2 の試行回数の制限と、2.4.3 の再利用の防止を行い、結果を "ok"・"invalid"・"locked" の3つで返す。3.4.3 の check_code() から呼ばれ、ログイン(3.4.4)と登録の確定(3.4.5)の両方で使う。
def verify(db, user, code):
"""照合:入力されたコードが正しいかを確かめる。
戻り値は次の3つのどれか。
"ok" : コードが正しく、まだ使われていない
"invalid" : コードが違う、または使用済み(再利用)
"locked" : 失敗が続いたためロック中
ログイン(/login/totp)と登録の確定(/totp/activate)の両方から呼ばれる。
"""
# 現在時刻(Unix 時刻)。この関数の中では、この1つの時刻を使い続ける
now = time.time()
# --- 1. ロック中なら、コードを見ずに断る ---
# totp_locked_until にはロックが解ける時刻が入っている(ロックしていなければ空)。
# その時刻がまだ来ていなければロック中。正しいコードが送られても通さない
if user["totp_locked_until"] and user["totp_locked_until"] > now:
return "locked"
# --- 2. コードが一致するかを調べる ---
# DB には暗号化した秘密鍵が入っているので、照合するときだけ復号して元に戻す
secret = fernet.decrypt(user["totp_secret"].encode()).decode()
step = find_matched_step(secret, code, now)
# --- 3. 一致したら、同じコードが2回使われていないかを確かめて記録する ---
if step is not None:
# 「保存済みのステップより新しいときだけ」ステップ番号を更新する。
# 条件(WHERE の後ろ)を UPDATE 文の中に入れているのがポイント。
# - 初めて使うステップなら、条件に合うので1行更新される → 成功
# - 同じコードの2回目なら、条件に合わないので0行しか更新されない → 失敗
# 「SELECT で読んで比べてから UPDATE する」と2回に分けると、
# 同じコードを同時に2回送られたときに、両方とも「まだ使っていない」と読んで通ってしまう。
# 1つの UPDATE 文にすれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。
# あわせて、成功したので失敗回数(totp_failures)を0に戻す
updated = db.execute(
"UPDATE users SET totp_last_used_step = ?, totp_failures = 0"
" WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?)",
(step, user["id"], step),
).rowcount # rowcount は、更新した行の数
if updated == 1:
return "ok"
# --- 4. ここに来たら失敗。失敗を数え、上限に達したらロックする ---
# コードが違った場合と、使用済みのコードだった場合の両方がここに来る。
# 回数は TOTP待ちトークンごとではなく、利用者ごとに DB に数える。
# トークンごとに数えると、ログインし直すだけで回数が0に戻ってしまうため
db.execute("UPDATE users SET totp_failures = totp_failures + 1 WHERE id = ?", (user["id"],))
# 増やしたあとの失敗回数を読み直す
failures = db.execute("SELECT totp_failures FROM users WHERE id = ?", (user["id"],)).fetchone()[0]
if failures >= MAX_FAILURES:
# 上限に達したら、ロックが解ける時刻(今から15分後)を記録する。
# 失敗回数は0に戻し、ロックが解けたあと、また5回試せるようにする
db.execute(
"UPDATE users SET totp_failures = 0, totp_locked_until = ? WHERE id = ?",
(now + LOCK_SECONDS, user["id"]),
)
return "locked"
return "invalid"
照合は3つの段階に分かれる。
3.3.4.1 ロック中なら、照合しない
ロック中は、正しいコードでも "locked" を返す。ロックが解ける時刻と今の時刻を比べるだけで、コードは見ない。
3.3.4.2 同じコードを2回使えなくする
コードが一致したら、そのステップ番号を totp_last_used_step に保存する。次からは、保存したステップ以下のコードを拒否する。
ポイントは、「保存済みのステップより新しいときだけ更新する」という条件を、UPDATE 文の WHERE に入れていることである。
UPDATE users SET totp_last_used_step = ?, totp_failures = 0
WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?)
同じコードの2回目は、WHERE の条件に合わないので、更新した行が0件になる。rowcount が1のときだけ成功とする。
「SELECT で読んで比べてから、UPDATE で書く」と2回に分けると、同じコードを同時に2回送られたときに、両方が「まだ使っていない」と読んでしまい、両方とも通る。条件を UPDATE 文の中に入れれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。
3.3.4.3 失敗を数え、上限に達したらロックする
失敗したら totp_failures を1増やし、5回に達したら15分後の時刻を totp_locked_until に入れる。ロックしたら回数は0に戻し、解除後にまた5回試せるようにする。
回数は、DB の users テーブルに利用者ごとに数える。TOTP待ちトークンごとに数えると、ログインし直してTOTP待ちトークンを取り直すだけで、回数が0に戻ってしまう。
再利用(3.3.4.2)も、失敗として数える。
<details><summary>補足:なぜ秘密鍵はハッシュ化ではなく暗号化なのか</summary>
パスワードは、ハッシュ化(元に戻せない変換)して保存する。照合のときは、入力をハッシュ化して比べれば済むからである。
TOTP の秘密鍵は、ハッシュ化できない。照合のたびに、サーバーが元の秘密鍵でコードを計算する必要があるからである。このため、元に戻せる暗号化で保存し、照合するときだけ復号する(RFC 6238 §5.1)。
</details>
<details><summary>totp.py の全体</summary>
"""TOTP の登録と照合を行う部品。
main.py(API)から呼ばれる。このファイルには、次の3つの関数がある。
- enroll() : 登録。秘密鍵を作って DB に保存し、認証アプリに読ませる URI を返す
- find_matched_step() : 入力されたコードが、どのタイムステップのコードと一致するかを調べる
- verify() : 照合。コードが正しいかを確かめ、再利用の防止と試行回数の制限も行う
"""
# 環境変数を読むために使う(Python 標準)
import os
# 現在時刻(Unix 時刻)を得るために使う(Python 標準)
import time
# TOTP の秘密鍵の作成と、コードの計算を行うライブラリ
import pyotp
# 秘密鍵の暗号化と復号に使う仕組み(記事の 2.4.4)
from cryptography.fernet import Fernet
# ---------------------------------------------------------------------------
# 設計値(記事の 3.1.5 で決めた値)
# ---------------------------------------------------------------------------
# 認証アプリに表示されるサービス名。アプリには「TOTP-Demo(alice@example.com)」のように出る
ISSUER = "TOTP-Demo"
# 時刻のずれを、前後何ステップまで許すか。
# 1 なら「1つ前」「現在」「1つ後」の3つのステップのコードを受け付ける(1ステップ=30秒)
VALID_WINDOW = 1
# 何回続けて失敗したらロックするか
MAX_FAILURES = 5
# ロックする時間(秒)。15分 × 60秒 = 900秒
LOCK_SECONDS = 15 * 60
# ---------------------------------------------------------------------------
# 暗号化の準備
# ---------------------------------------------------------------------------
# 秘密鍵を DB に保存するときに使う、暗号化の鍵を読み込む。
# 鍵はソースコードにも DB にも書かず、起動するときに環境変数 TOTP_ENCRYPTION_KEY で渡す。
# 環境変数が無いと、ここで KeyError になり、サーバーは起動しない。
fernet = Fernet(os.environ["TOTP_ENCRYPTION_KEY"])
def enroll(db, user):
"""登録:秘密鍵を作り、暗号化して DB に保存する。
戻り値は otpauth:// で始まる URI(文字列)。
main.py がこれを QR コードにして返し、利用者が認証アプリで読み取る。
この関数では、まだ TOTP を有効にしない(有効にするのは /totp/activate)。
"""
# 利用者ごとの秘密鍵を、ランダムに作る。
# 160ビットの値を Base32(英大文字 A〜Z と数字 2〜7)の32文字で表したもの
secret = pyotp.random_base32()
# 秘密鍵を暗号化して、DB に保存する。
# - fernet.encrypt() は bytes を受け取って bytes を返すので、
# .encode() で文字列 → bytes に、.decode() で bytes → 文字列に変換している
# - totp_last_used_step(最後に使ったステップ)を空に戻すのは、
# 秘密鍵を作り直したときに、前の秘密鍵の記録が残らないようにするため
# - SQL の ? には、後ろのタプルの値が順に入る(SQL インジェクションを防ぐ書き方)
db.execute(
"UPDATE users SET totp_secret = ?, totp_last_used_step = NULL WHERE id = ?",
(fernet.encrypt(secret.encode()).decode(), user["id"]),
)
# 認証アプリに渡す otpauth:// URI を作って返す。
# name はアカウント名(ここではメールアドレス)、issuer_name はサービス名。
# 暗号化する前の秘密鍵が URI に入るので、この URI(と QR)は人に見せてはいけない
return pyotp.TOTP(secret).provisioning_uri(name=user["email"], issuer_name=ISSUER)
def find_matched_step(secret, code, now):
"""入力されたコードが、どのタイムステップのコードと一致するかを調べる。
引数
secret : 秘密鍵(復号した Base32 の文字列)
code : 利用者が入力した6桁のコード(文字列)
now : 現在時刻(Unix 時刻。秒)
戻り値
一致したタイムステップ番号(整数)。どれとも一致しなければ None。
pyotp には verify() という照合用の関数もあるが、True / False しか返さず、
どのステップで一致したかが分からない。再利用を防ぐにはステップ番号が要るので、自分で調べる。
"""
# この秘密鍵でコードを計算する道具を作る(既定値は SHA-1・6桁・30秒)
totp = pyotp.TOTP(secret)
# 現在のタイムステップ番号を求める。
# T = floor(Unix 時刻 / 30)。// は「割り算の商の整数部分」を求める演算子
current_step = int(now) // totp.interval
# offset を -1, 0, +1 と変えながら、「1つ前」「現在」「1つ後」のステップのコードと比べる
for offset in range(-VALID_WINDOW, VALID_WINDOW + 1):
# totp.at(now, offset) は、現在から offset ステップずらした時刻のコードを返す。
# 比べるときは == ではなく strings_equal を使う。
# == は先頭から比べて、違う文字が見つかった時点で止まるので、
# かかった時間の差から「何文字目まで合っていたか」を推測される余地がある。
# strings_equal は、一致の度合いにかかわらず同じ時間で比べる
if pyotp.utils.strings_equal(code, totp.at(now, offset)):
return current_step + offset
# どのステップのコードとも一致しなかった
return None
def verify(db, user, code):
"""照合:入力されたコードが正しいかを確かめる。
戻り値は次の3つのどれか。
"ok" : コードが正しく、まだ使われていない
"invalid" : コードが違う、または使用済み(再利用)
"locked" : 失敗が続いたためロック中
ログイン(/login/totp)と登録の確定(/totp/activate)の両方から呼ばれる。
"""
# 現在時刻(Unix 時刻)。この関数の中では、この1つの時刻を使い続ける
now = time.time()
# --- 1. ロック中なら、コードを見ずに断る ---
# totp_locked_until にはロックが解ける時刻が入っている(ロックしていなければ空)。
# その時刻がまだ来ていなければロック中。正しいコードが送られても通さない
if user["totp_locked_until"] and user["totp_locked_until"] > now:
return "locked"
# --- 2. コードが一致するかを調べる ---
# DB には暗号化した秘密鍵が入っているので、照合するときだけ復号して元に戻す
secret = fernet.decrypt(user["totp_secret"].encode()).decode()
step = find_matched_step(secret, code, now)
# --- 3. 一致したら、同じコードが2回使われていないかを確かめて記録する ---
if step is not None:
# 「保存済みのステップより新しいときだけ」ステップ番号を更新する。
# 条件(WHERE の後ろ)を UPDATE 文の中に入れているのがポイント。
# - 初めて使うステップなら、条件に合うので1行更新される → 成功
# - 同じコードの2回目なら、条件に合わないので0行しか更新されない → 失敗
# 「SELECT で読んで比べてから UPDATE する」と2回に分けると、
# 同じコードを同時に2回送られたときに、両方とも「まだ使っていない」と読んで通ってしまう。
# 1つの UPDATE 文にすれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。
# あわせて、成功したので失敗回数(totp_failures)を0に戻す
updated = db.execute(
"UPDATE users SET totp_last_used_step = ?, totp_failures = 0"
" WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?)",
(step, user["id"], step),
).rowcount # rowcount は、更新した行の数
if updated == 1:
return "ok"
# --- 4. ここに来たら失敗。失敗を数え、上限に達したらロックする ---
# コードが違った場合と、使用済みのコードだった場合の両方がここに来る。
# 回数は TOTP待ちトークンごとではなく、利用者ごとに DB に数える。
# トークンごとに数えると、ログインし直すだけで回数が0に戻ってしまうため
db.execute("UPDATE users SET totp_failures = totp_failures + 1 WHERE id = ?", (user["id"],))
# 増やしたあとの失敗回数を読み直す
failures = db.execute("SELECT totp_failures FROM users WHERE id = ?", (user["id"],)).fetchone()[0]
if failures >= MAX_FAILURES:
# 上限に達したら、ロックが解ける時刻(今から15分後)を記録する。
# 失敗回数は0に戻し、ロックが解けたあと、また5回試せるようにする
db.execute(
"UPDATE users SET totp_failures = 0, totp_locked_until = ? WHERE id = ?",
(now + LOCK_SECONDS, user["id"]),
)
return "locked"
return "invalid"
</details>
3.4 APIを書く(main.py)
本節では3.3 の totp.py を使って、3.1.3 のエンドポイントを持つ API を main.py に書く。次の順に組み立てる。
| 節 | 書くもの | 対応する説明 |
|---|---|---|
| 3.4.1 | 読み込むライブラリと、トークンに署名する鍵 | 2.5(2種類のトークン) |
| 3.4.2 | DB と、動作確認用の利用者 | 3.1.4(users テーブル) |
| 3.4.3 | リクエストの形と、エンドポイントで共通に使う関数 | ― |
| 3.4.4 | ログイン:/login、/login/totp、/me | 3.1.2(ログインの流れ) |
| 3.4.5 | 登録:/totp/enroll、/totp/activate | 2.3(秘密鍵の受け渡し) |
~/totp-demo に main.py を作り、3.4.1〜3.4.5 のコードを順に書いていく。
3.4.1 JWTの鍵を2つに分ける
main.py の先頭では、使うライブラリと 3.3 の totp.py を読み込み、ログインで発行する2種類のトークン(2.5 のアクセストークンと TOTP待ちトークン)に署名するための鍵を用意する。
"""ID・パスワードのログインに、TOTP の確認を2段目として足した API。
起動すると、次の5つのエンドポイント(URL)が使えるようになる。
- POST /login : ログインの1段目。ID・パスワードを確かめる
- POST /login/totp : ログインの2段目。認証アプリのコード(TOTP)を確かめる
- GET /me : ログイン中の利用者を返す(ログインできたかの確認用)
- POST /totp/enroll : TOTP の登録。秘密鍵を作り、QR コードを返す
- POST /totp/activate : 登録の確定。認証アプリのコードが正しければ TOTP を有効にする
"""
# --- Python 標準のライブラリ ---
# 推測されにくいランダムな値を作る。パスワードを安全に比べるのにも使う
import secrets
# SQLite(ファイル1つで動くデータベース)を使う
import sqlite3
# 現在時刻(Unix 時刻)を得る。トークンの有効期限の計算に使う
import time
# --- 3.2.2 で入れたライブラリ ---
# JWT(署名付きのトークン)を作ったり、確かめたりする
import jwt
# otpauth:// URI を QR コードの画像にする
import qrcode
import qrcode.image.svg
# API を作るためのフレームワーク
from fastapi import Depends, FastAPI, HTTPException, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
# リクエストで受け取る JSON の形を定義する
from pydantic import BaseModel
# --- 3.3 で作った TOTP の部品 ---
import totp
# ---------------------------------------------------------------------------
# トークンに署名する鍵
# ---------------------------------------------------------------------------
# ログインで発行する2種類のトークンに、それぞれ別の鍵で署名する。
# - ACCESS_KEY : アクセストークン(ログイン済みの証明)用
# - PENDING_KEY : TOTP待ちトークン(パスワードは通ったが TOTP はまだ、という証明)用
# 同じ鍵にすると、TOTP待ちトークンがアクセストークンの確認を通ってしまい、
# TOTP を入れずに API を使えてしまう。鍵を分ければ、それを防げる。
# 鍵は起動のたびに作り直す(デモなので、再起動したらログインし直せばよい)。
# token_hex(32) は、32バイトのランダムな値を16進数の文字列(64文字)で返す
ACCESS_KEY = secrets.token_hex(32)
PENDING_KEY = secrets.token_hex(32)
アクセストークン用(ACCESS_KEY)とTOTP待ちトークン用(PENDING_KEY)で、署名の鍵を分けている。
TOTP待ちトークンは「パスワードは通ったが、TOTP はまだ」という状態を表す。もしアクセストークンと同じ鍵で署名すると、TOTP待ちトークンも署名の確認を通ってしまい、TOTP を飛ばして API を使えてしまう。鍵を分けておけば、TOTP待ちトークンはアクセストークンの確認(3.4.3)で必ず失敗する。
鍵は起動のたびに secrets.token_hex() で作り直している。デモなので再起動したらログインし直せばよい。
3.4.2 DBと動作確認用の利用者
次に、3.1.4 の users テーブルを SQLite に作り、4.1 の動作確認で使う利用者を1人入れておく。
# ---------------------------------------------------------------------------
# データベース(users テーブル)
# ---------------------------------------------------------------------------
# demo.db というファイルの DB を開く(無ければ作る)。ファイルは、起動したディレクトリにできる。
# - isolation_level=None : SQL を1文実行するたびに、自動で確定(コミット)する
# - check_same_thread=False : FastAPI は複数のスレッドから DB を使うので、それを許す
db = sqlite3.connect("demo.db", isolation_level=None, check_same_thread=False)
# 検索結果を user["email"] のように、列の名前で取り出せるようにする
db.row_factory = sqlite3.Row
# users テーブルを作る(すでにあれば何もしない)。
# id・email・password が元からあるログインの列、totp_ で始まる5列が TOTP のために足した列
db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
email TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
totp_secret TEXT, -- 秘密鍵(暗号化して保存)
totp_enabled INTEGER NOT NULL DEFAULT 0, -- 1 なら TOTP を求める
totp_last_used_step INTEGER, -- 最後に成功したタイムステップ番号
totp_failures INTEGER NOT NULL DEFAULT 0,-- 続けて失敗した回数
totp_locked_until REAL -- ロックが解ける時刻(Unix 時刻)
)""")
# 動作確認用の利用者 alice を入れる(すでにいれば何もしない)。
# パスワードは、デモのため平文で保存している。実際のシステムではハッシュ化して保存する
db.execute("INSERT OR IGNORE INTO users (email, password) VALUES ('alice@example.com', 'password')")
# API の本体を作る。このあとの @app.post(...) などで、エンドポイントを足していく
app = FastAPI()
- 起動すると demo.db ができ、users テーブル(3.1.4)と、動作確認用の利用者 alice が入る
- isolation_level=None で、SQL を1文ずつ自動で確定(コミット)させている
- パスワードは、デモのため平文で保存している(4.2.1)
3.4.3 リクエストの形と共通の処理
各エンドポイントで共通して使う部品を用意する。前半(3/6)は、リクエストで受け取る JSON の形の定義である。後半(4/6)は、トークンを発行・確認する関数と、3.3.4 の照合結果を HTTP の応答に変える関数である。
# ---------------------------------------------------------------------------
# リクエストで受け取る JSON の形
# ---------------------------------------------------------------------------
# FastAPI は、届いた JSON がこの形に合っているかを自動で確かめる。
# 項目が足りないなどの場合は、こちらの処理に来る前に 422 エラーを返す
class LoginRequest(BaseModel):
"""/login で受け取る JSON。例:{"email": "alice@example.com", "password": "password"}"""
email: str
password: str
class TotpLoginRequest(BaseModel):
"""/login/totp で受け取る JSON。/login で受け取った TOTP待ちトークンと、6桁のコード"""
totp_token: str
code: str
class CodeRequest(BaseModel):
"""/totp/activate で受け取る JSON。認証アプリに表示された6桁のコード"""
code: str
# ---------------------------------------------------------------------------
# エンドポイントで共通して使う関数
# ---------------------------------------------------------------------------
def make_token(email, key, minutes):
"""トークン(JWT)を作る。
中身は「誰か(sub:メールアドレス)」と「いつまで有効か(exp:Unix 時刻)」の2つ。
渡された鍵で署名するので、鍵を知らない人には同じトークンを作れない。
"""
return jwt.encode({"sub": email, "exp": int(time.time()) + minutes * 60}, key, algorithm="HS256")
def find_user(token, key):
"""トークンを確かめて、その利用者を DB から取り出す。
渡された鍵で署名を確かめる。署名が合わない(鍵が違う)、期限が切れている、
中身が壊れている場合は、401(認証エラー)を返して処理を止める。
"""
try:
email = jwt.decode(token, key, algorithms=["HS256"])["sub"]
except jwt.PyJWTError:
raise HTTPException(401, "Invalid token")
return db.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
def current_user(cred: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
"""ログイン中の利用者を返す。ログインが必要なエンドポイントで使う。
リクエストのヘッダー「Authorization: Bearer <トークン>」からトークンを取り出し、
アクセストークン用の鍵(ACCESS_KEY)で確かめる。
TOTP待ちトークンは PENDING_KEY で署名しているので、ここでは必ず 401 になる。
"""
return find_user(cred.credentials, ACCESS_KEY)
def check_code(user, code):
"""TOTP のコードを照合し、失敗なら HTTP のエラーを返して処理を止める。
totp.verify() の結果を、HTTP の応答に変える。
- "locked" → 429(Too Many Requests:試しすぎ)
- "invalid" → 401(認証エラー)
- "ok" → 何もせずに戻る(呼び出し元の処理が続く)
"""
result = totp.verify(db, user, code)
if result == "locked":
raise HTTPException(429, "Too many failed attempts")
if result == "invalid":
raise HTTPException(401, "Invalid code")
- make_token() は、メールアドレスと有効期限を入れたトークンを、渡された鍵で署名する
- find_user() は、渡された鍵で署名を確かめ、利用者を返す。鍵が違えば 401 にする
- current_user() は、アクセストークンの鍵(ACCESS_KEY)でしか確かめない。このため、TOTP待ちトークンを送ると 401 になる
- check_code() は、3.3.4 の結果を HTTP の応答に変える。ロック中は 429、コードが違えば 401 にする
3.4.4 ログインを2段階にする
3.1.2 の図のログインの流れ(工程1〜7)を、/login(工程1〜4)と /login/totp(工程5〜7)の2つのエンドポイントで実装する。あわせて、ログインできたかを確かめるための /me も作る。
# ---------------------------------------------------------------------------
# ログイン
# ---------------------------------------------------------------------------
@app.post("/login")
def login(req: LoginRequest):
"""ログインの1段目:ID・パスワードを確かめる(3.1.2 の図の工程1〜4)。
- TOTP を登録していない利用者 → アクセストークンを返す(ログイン完了)
- TOTP を登録済みの利用者 → TOTP待ちトークン(5分)を返す(2段目へ進む)
"""
# メールアドレスで利用者を探す
user = db.execute("SELECT * FROM users WHERE email = ?", (req.email,)).fetchone()
# 利用者がいない、またはパスワードが違えば 401。
# どちらの場合も同じメッセージにして、「そのメールアドレスが登録されているか」を教えないようにする。
# compare_digest は、一致の度合いにかかわらず同じ時間で比べる関数
if user is None or not secrets.compare_digest(req.password, user["password"]):
raise HTTPException(401, "Invalid email or password")
# TOTP を登録済みなら、アクセストークンの代わりに TOTP待ちトークン(有効期限5分)を返す
if user["totp_enabled"]:
return {"totp_token": make_token(user["email"], PENDING_KEY, 5)}
# TOTP を登録していなければ、これまでどおりアクセストークン(有効期限30分)を返す
return {"access_token": make_token(user["email"], ACCESS_KEY, 30)}
@app.post("/login/totp")
def login_totp(req: TotpLoginRequest):
"""ログインの2段目:認証アプリのコードを確かめる(3.1.2 の図の工程5〜7)。
TOTP待ちトークンとコードが正しければ、アクセストークンを返す。
"""
# TOTP待ちトークンを、TOTP待ちトークン用の鍵(PENDING_KEY)で確かめ、利用者を取り出す
user = find_user(req.totp_token, PENDING_KEY)
# コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる
check_code(user, req.code)
# 照合に成功したので、アクセストークン(有効期限30分)を返す
return {"access_token": make_token(user["email"], ACCESS_KEY, 30)}
@app.get("/me")
def me(user=Depends(current_user)):
"""ログイン中の利用者を返す。アクセストークンが無いと 401 になる。
Depends(current_user) と書くと、この関数が呼ばれる前に current_user() が実行され、
その戻り値(ログイン中の利用者)が user に入る。
"""
return {"email": user["email"], "totp_enabled": bool(user["totp_enabled"])}
- /login は、パスワードが正しいとき、TOTP が有効ならTOTP待ちトークン(5分)を、無効ならアクセストークンを返す。TOTP を使わない利用者の動きは、今までと変わらない
- /login/totp は、TOTP待ちトークンを PENDING_KEY で確かめてから、コードを照合する
- /me は、アクセストークンでしか呼べない。4.1 で、TOTP待ちトークンがアクセストークンとして通らないことを確かめるのに使う
3.4.5 登録と有効化
最後に、TOTP を登録するためのエンドポイントを作る。2.3.1 の登録の手順のうち、秘密鍵を作って QR コードを返す部分(手順1〜2)を /totp/enroll が、認証アプリのコードを確かめて登録を確定する部分(手順4)を /totp/activate が受け持つ。
# ---------------------------------------------------------------------------
# TOTP の登録
# ---------------------------------------------------------------------------
@app.post("/totp/enroll")
def enroll(user=Depends(current_user)):
"""TOTP の登録:秘密鍵を作り、QR コード(SVG 画像)を返す(2.3.1 の手順1〜2)。
ログインした本人が使う(アクセストークンが要る)。
この時点では、まだ TOTP は有効にならない。/totp/activate でコードが通ってから有効になる。
"""
# すでに有効なら、登録し直させない(秘密鍵が勝手に作り直されないようにする)。409 は「状態が合わない」
if user["totp_enabled"]:
raise HTTPException(409, "Already enabled")
# 秘密鍵を作って DB に保存し、otpauth:// URI を受け取る(totp.py の enroll())
uri = totp.enroll(db, user)
# URI を QR コードにする。QR はサーバーの中で作る。
# 外部の QR 生成サービスに URI を送ると、秘密鍵が外に漏れるため。
# SvgPathFillImage は、白い背景付きの SVG 画像を作る
svg = qrcode.make(uri, image_factory=qrcode.image.svg.SvgPathFillImage).to_string()
# SVG 画像として返す。curl の -o で保存し、ブラウザなどで開いて認証アプリで読み取る
return Response(svg, media_type="image/svg+xml")
@app.post("/totp/activate")
def activate(req: CodeRequest, user=Depends(current_user)):
"""登録の確定:認証アプリのコードが正しければ、TOTP を有効にする(2.3.1 の手順4)。
QR を正しく読み取れたか(サーバーと認証アプリが同じ秘密鍵を持てたか)を確かめるためのもので、
ログインではない。
"""
# まだ /totp/enroll で秘密鍵を作っていなければ、確かめようがないので 400(リクエストの誤り)
if user["totp_secret"] is None:
raise HTTPException(400, "Call /totp/enroll first")
# コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる
check_code(user, req.code)
# 照合に成功したので TOTP を有効にする。以後のログインでは2段目(/login/totp)を求める
db.execute("UPDATE users SET totp_enabled = 1 WHERE id = ?", (user["id"],))
return {"totp_enabled": True}
- /totp/enroll は、秘密鍵を作り、QR コードを SVG で返す。QR はサーバーの中で作る(外部の QR 生成サービスに送ると、秘密鍵が漏れる)
- /totp/activate は、認証アプリに出たコードが通ったら、TOTP を有効にする
<details><summary>補足:なぜ確認してから有効にするのか</summary>
QR を読み取った時点で有効にすると、読み取りに失敗していた場合や、別の端末で読み取ってしまった場合に、本人がログインできなくなる。認証アプリに出たコードを1回入れてもらい、通ってから有効にすれば、この事故を防げる。
確認に使ったコードは、3.3.4.2 で使用済みになる。このため、確認と同じ30秒の間は、そのコードでログインできない。
</details>
<details><summary>main.py の全体</summary>
"""ID・パスワードのログインに、TOTP の確認を2段目として足した API。
起動すると、次の5つのエンドポイント(URL)が使えるようになる。
- POST /login : ログインの1段目。ID・パスワードを確かめる
- POST /login/totp : ログインの2段目。認証アプリのコード(TOTP)を確かめる
- GET /me : ログイン中の利用者を返す(ログインできたかの確認用)
- POST /totp/enroll : TOTP の登録。秘密鍵を作り、QR コードを返す
- POST /totp/activate : 登録の確定。認証アプリのコードが正しければ TOTP を有効にする
"""
--- Python 標準のライブラリ ---
推測されにくいランダムな値を作る。パスワードを安全に比べるのにも使う
import secrets
SQLite(ファイル1つで動くデータベース)を使う
import sqlite3
現在時刻(Unix 時刻)を得る。トークンの有効期限の計算に使う
import time
--- 3.2.2 で入れたライブラリ ---
JWT(署名付きのトークン)を作ったり、確かめたりする
import jwt
otpauth:// URI を QR コードの画像にする
import qrcode
import qrcode.image.svg
API を作るためのフレームワーク
from fastapi import Depends, FastAPI, HTTPException, Response
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
リクエストで受け取る JSON の形を定義する
from pydantic import BaseModel
--- 3.3 で作った TOTP の部品 ---
import totp
---------------------------------------------------------------------------
トークンに署名する鍵
---------------------------------------------------------------------------
ログインで発行する2種類のトークンに、それぞれ別の鍵で署名する。
- ACCESS_KEY : アクセストークン(ログイン済みの証明)用
- PENDING_KEY : TOTP待ちトークン(パスワードは通ったが TOTP はまだ、という証明)用
同じ鍵にすると、TOTP待ちトークンがアクセストークンの確認を通ってしまい、
TOTP を入れずに API を使えてしまう。鍵を分ければ、それを防げる。
鍵は起動のたびに作り直す(デモなので、再起動したらログインし直せばよい)。
token_hex(32) は、32バイトのランダムな値を16進数の文字列(64文字)で返す
ACCESS_KEY = secrets.token_hex(32)
PENDING_KEY = secrets.token_hex(32)
---------------------------------------------------------------------------
データベース(users テーブル)
---------------------------------------------------------------------------
demo.db というファイルの DB を開く(無ければ作る)。ファイルは、起動したディレクトリにできる。
- isolation_level=None : SQL を1文実行するたびに、自動で確定(コミット)する
- check_same_thread=False : FastAPI は複数のスレッドから DB を使うので、それを許す
db = sqlite3.connect("demo.db", isolation_level=None, check_same_thread=False)
検索結果を user["email"] のように、列の名前で取り出せるようにする
db.row_factory = sqlite3.Row
users テーブルを作る(すでにあれば何もしない)。
id・email・password が元からあるログインの列、totp_ で始まる5列が TOTP のために足した列
db.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
email TEXT UNIQUE NOT NULL,
password TEXT NOT NULL,
totp_secret TEXT, -- 秘密鍵(暗号化して保存)
totp_enabled INTEGER NOT NULL DEFAULT 0, -- 1 なら TOTP を求める
totp_last_used_step INTEGER, -- 最後に成功したタイムステップ番号
totp_failures INTEGER NOT NULL DEFAULT 0,-- 続けて失敗した回数
totp_locked_until REAL -- ロックが解ける時刻(Unix 時刻)
)""")
動作確認用の利用者 alice を入れる(すでにいれば何もしない)。
パスワードは、デモのため平文で保存している。実際のシステムではハッシュ化して保存する
db.execute("INSERT OR IGNORE INTO users (email, password) VALUES ('alice@example.com', 'password')")
API の本体を作る。このあとの @app.post(...) などで、エンドポイントを足していく
app = FastAPI()
---------------------------------------------------------------------------
リクエストで受け取る JSON の形
---------------------------------------------------------------------------
FastAPI は、届いた JSON がこの形に合っているかを自動で確かめる。
項目が足りないなどの場合は、こちらの処理に来る前に 422 エラーを返す
class LoginRequest(BaseModel):
"""/login で受け取る JSON。例:{"email": "alice@example.com", "password": "password"}"""
email: str
password: str
class TotpLoginRequest(BaseModel):
"""/login/totp で受け取る JSON。/login で受け取った TOTP待ちトークンと、6桁のコード"""
totp_token: str
code: str
class CodeRequest(BaseModel):
"""/totp/activate で受け取る JSON。認証アプリに表示された6桁のコード"""
code: str
---------------------------------------------------------------------------
エンドポイントで共通して使う関数
---------------------------------------------------------------------------
def make_token(email, key, minutes):
"""トークン(JWT)を作る。
中身は「誰か(sub:メールアドレス)」と「いつまで有効か(exp:Unix 時刻)」の2つ。
渡された鍵で署名するので、鍵を知らない人には同じトークンを作れない。
"""
return jwt.encode({"sub": email, "exp": int(time.time()) + minutes * 60}, key, algorithm="HS256")
def find_user(token, key):
"""トークンを確かめて、その利用者を DB から取り出す。
渡された鍵で署名を確かめる。署名が合わない(鍵が違う)、期限が切れている、
中身が壊れている場合は、401(認証エラー)を返して処理を止める。
"""
try:
email = jwt.decode(token, key, algorithms=["HS256"])["sub"]
except jwt.PyJWTError:
raise HTTPException(401, "Invalid token")
return db.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
def current_user(cred: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
"""ログイン中の利用者を返す。ログインが必要なエンドポイントで使う。
リクエストのヘッダー「Authorization: Bearer <トークン>」からトークンを取り出し、
アクセストークン用の鍵(ACCESS_KEY)で確かめる。
TOTP待ちトークンは PENDING_KEY で署名しているので、ここでは必ず 401 になる。
"""
return find_user(cred.credentials, ACCESS_KEY)
def check_code(user, code):
"""TOTP のコードを照合し、失敗なら HTTP のエラーを返して処理を止める。
totp.verify() の結果を、HTTP の応答に変える。
- "locked" → 429(Too Many Requests:試しすぎ)
- "invalid" → 401(認証エラー)
- "ok" → 何もせずに戻る(呼び出し元の処理が続く)
"""
result = totp.verify(db, user, code)
if result == "locked":
raise HTTPException(429, "Too many failed attempts")
if result == "invalid":
raise HTTPException(401, "Invalid code")
---------------------------------------------------------------------------
ログイン
---------------------------------------------------------------------------
@app.post("/login")
def login(req: LoginRequest):
"""ログインの1段目:ID・パスワードを確かめる(3.1.2 の図の工程1〜4)。
- TOTP を登録していない利用者 → アクセストークンを返す(ログイン完了)
- TOTP を登録済みの利用者 → TOTP待ちトークン(5分)を返す(2段目へ進む)
"""
# メールアドレスで利用者を探す
user = db.execute("SELECT * FROM users WHERE email = ?", (req.email,)).fetchone()
# 利用者がいない、またはパスワードが違えば 401。
# どちらの場合も同じメッセージにして、「そのメールアドレスが登録されているか」を教えないようにする。
# compare_digest は、一致の度合いにかかわらず同じ時間で比べる関数
if user is None or not secrets.compare_digest(req.password, user["password"]):
raise HTTPException(401, "Invalid email or password")
# TOTP を登録済みなら、アクセストークンの代わりに TOTP待ちトークン(有効期限5分)を返す
if user["totp_enabled"]:
return {"totp_token": make_token(user["email"], PENDING_KEY, 5)}
# TOTP を登録していなければ、これまでどおりアクセストークン(有効期限30分)を返す
return {"access_token": make_token(user["email"], ACCESS_KEY, 30)}
@app.post("/login/totp")
def login_totp(req: TotpLoginRequest):
"""ログインの2段目:認証アプリのコードを確かめる(3.1.2 の図の工程5〜7)。
TOTP待ちトークンとコードが正しければ、アクセストークンを返す。
"""
# TOTP待ちトークンを、TOTP待ちトークン用の鍵(PENDING_KEY)で確かめ、利用者を取り出す
user = find_user(req.totp_token, PENDING_KEY)
# コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる
check_code(user, req.code)
# 照合に成功したので、アクセストークン(有効期限30分)を返す
return {"access_token": make_token(user["email"], ACCESS_KEY, 30)}
@app.get("/me")
def me(user=Depends(current_user)):
"""ログイン中の利用者を返す。アクセストークンが無いと 401 になる。
Depends(current_user) と書くと、この関数が呼ばれる前に current_user() が実行され、
その戻り値(ログイン中の利用者)が user に入る。
"""
return {"email": user["email"], "totp_enabled": bool(user["totp_enabled"])}
---------------------------------------------------------------------------
TOTP の登録
---------------------------------------------------------------------------
@app.post("/totp/enroll")
def enroll(user=Depends(current_user)):
"""TOTP の登録:秘密鍵を作り、QR コード(SVG 画像)を返す(2.3.1 の手順1〜2)。
ログインした本人が使う(アクセストークンが要る)。
この時点では、まだ TOTP は有効にならない。/totp/activate でコードが通ってから有効になる。
"""
# すでに有効なら、登録し直させない(秘密鍵が勝手に作り直されないようにする)。409 は「状態が合わない」
if user["totp_enabled"]:
raise HTTPException(409, "Already enabled")
# 秘密鍵を作って DB に保存し、otpauth:// URI を受け取る(totp.py の enroll())
uri = totp.enroll(db, user)
# URI を QR コードにする。QR はサーバーの中で作る。
# 外部の QR 生成サービスに URI を送ると、秘密鍵が外に漏れるため。
# SvgPathFillImage は、白い背景付きの SVG 画像を作る
svg = qrcode.make(uri, image_factory=qrcode.image.svg.SvgPathFillImage).to_string()
# SVG 画像として返す。curl の -o で保
Mentions
TOTPを実装する 0. はじめに この記事は、学んだ技術や知識についてを備忘録としてまとめることを目的にしている。今回はTOTPについてまとめる。 > 注意 > 本記事のコードは、仕組みを確かめるための検証用です。パスワードの扱いなど、実際のシステムに組み込むときに必要な処理を省いています。実際のシステムで使う場合は、4.2 を読んでください。 1. 開発環境 <!-- TODO: Mac で確認して差し替える --> - macOS - uv 0.7.8 - Python 3.13.5(uv で用意する) - FastAPI 0.162.0、uvicorn 0.38.0 - PyJWT 2.10.1 - pyotp 2.10.0 - qrcode 8.2 - cryptography 46.0.3 - SQLite(Python 標準の sqlite3 と、macOS に入っている sqlite3 コマンド) - スマートフォンの認証アプリ(Google Authenticator、Microsoft Authenticator など) 2. 基本概念 2.1 多要素認証 本人を確かめる手がかり(認証の要素)は、次の3種類に分けられる。 | 要素 | 何で確かめるか | 例 | |---|---|---| | 知識要素 | 本人だけが知っていること | パスワード、暗証番号 | | 所持要素 | 本人だけが持っているもの | スマートフォン、IC カード | | 生体要素 | 本人の体の特徴 | 指紋、顔 | 種類の違う要素を2つ以上組み合わせる認証を、多要素認証(MFA)という。 ID・パスワードだけのログインは知識要素1つに頼っている。パスワードは漏えいや使い回し、推測によって他人に知られると、それだけでログインされてしまう。そこで知識要素に所持要素や生体要素を足し、パスワードを知られただけではログインできないようにする。 所持要素を足す場合は、利用者が端末を持っていることをネットワーク越しにサーバーへ示す必要がある。その方法の1つが端末に使い捨てのコードを作らせるワンタイムパスワードである。 2.2 ワンタイムパスワード(OTP) 2.2.1 使い捨てのコード 所持要素を確かめるには、その端末にしか作れない値を入力してもらえばよい。ただしその値が毎回同じだと、パスワードと同じく一度知られただけで使われ続けてしまう。そこで値をログインのたびに変え、一度使った値は使えないようにする。これがワンタイムパスワード(OTP)である。 OTP(One-Time Password) は、次の3つの手順で「端末を持っていること」を確かめる。 1. 端末とサーバーが同じ秘密の値(秘密鍵)をあらかじめ共有しておく 2. 端末は秘密鍵と「その時だけの値」からコードを計算して表示し、利用者がそれを入力する 3. サーバーも同じ秘密鍵と同じ「その時だけの値」から同じ計算をし、入力されたコードと比べる 秘密鍵を持たない端末は正しいコードを作れない。このため、コードが一致すれば利用者は秘密鍵の入った端末を手元に持っていると判断できる。またコードはその時だけのものなので、盗み見られてもあとで使い回すことはできない(有効な間に使い回されるのを防ぐ方法は2.4.3 で説明する)。 「その時だけの値」に何を使うかで方式が分かれる。 回数(カウンター)を使うのが HOTP(RFC 4226)、 時刻を使うのが TOTP(RFC 6238)である。 2.2.2 HOTP:秘密鍵とカウンターから作る TOTP は HOTP を元にしている。HOTP(HMAC-based One-Time Password)とは、秘密鍵 K と、サーバーと端末がそれぞれ数えるカウンター C からコードを作る方式(RFC 4226 §5.2)。 HOTP(K, C) = Truncate(HMAC-SHA-1(K, C)) HOTP のコードは、次の手順で作る。 1. HMAC-SHA-1(K, C):秘密鍵 K とカウンター C から、20バイトの値を作る 2. Truncate:20バイトのうち、最後の1バイトの下位4ビットが示す位置から4バイトを取り出し、先頭の1ビットを落として31ビットの数にする(RFC 4226 §5.3) 3. その数を $10^6$ で割った余りが、6桁のコードになる。桁数は最低6桁と決められている(RFC 4226 §5.3) sequenceDiagram participant A as 認証アプリ(端末) participant U as 利用者 participant S as サーバー Note over A,S: 両方が同じ秘密鍵 K と、同じ値のカウンター C を持っている A->>A: 端末のカウンター C で HOTP(K, C) を計算 A->>A: 端末のカウンター C を1つ進める A->>U: 6桁のコードを表示 U->>S: コードを入力して送る S->>S: サーバーのカウンター C で HOTP(K, C) を計算 S->>S: 入力されたコードと比べる S->>S: 一致したら、サーバーのカウンター C を1つ進める HOTP では、端末はコードを作るたびにカウンターを1つ進め、サーバーはコードの照合に成功するたびにカウンターを1つ進める。このため、端末でコードを作っただけで使わなかった場合、端末のカウンターだけが進み、サーバーと端末のカウンターがずれる。ずれると、端末が表示したコードとサーバーが計算したコードが一致せず、認証できなくなる。 RFC 4226 は、サーバーが先の数個のカウンターまで試し、一致したらカウンターを合わせ直す方法を定めている(RFC 4226 §7.4)。ただし、試す範囲を超えてずれると、合わせ直せない。 そこでカウンターの代わりにサーバーと端末が同じ値を自然に持てる時刻を用いた方式が考案された。それが TOTP である。 <details><summary>用語解説:HMACとは</summary> HMAC(Hash-based Message Authentication Code)は、秘密鍵とデータからハッシュ関数を使って決まった長さの値を作る計算である(RFC 2104 §2)。次の2つの性質がある。 - 同じ鍵と同じデータからは必ず同じ値になる - 鍵を知らなければその値を計算できない HMAC-SHA-1 はハッシュ関数に SHA-1 を使う HMAC で、結果は20バイトになる。HOTP はこの性質を使って、秘密鍵を持つ端末とサーバーだけが同じコードを作れるようにしている。 </details> 2.2.3 TOTP:カウンターを時刻に置き換える TOTP(Time-based One-Time Password)は、HOTP のカウンター C を時刻から求めた番号 T に置き換えたもの(RFC 6238 §4.2)。 T = floor(Unix時刻 / 30) TOTP(K) = HOTP(K, T) T は時刻を30秒ごとに区切った通し番号(タイムステップ)である。同じ30秒の間は T が同じなので、同じコードになる。コマンドを実行した時点(現在時刻)の T は、次のコマンドで確かめられる。 python3 -c "import time; print(int(time.time()) // 30)" サーバーと端末はそれぞれ自分の時計から T を求め、それぞれでコードを計算する。 sequenceDiagram participant A as 認証アプリ(端末) participant U as 利用者 participant S as サーバー Note over A,S: 両方が同じ秘密鍵 K を持っている A->>A: 端末の現在時刻で TOTP(K) を計算 A->>U: 6桁のコードを表示 U->>S: コードを入力して送る S->>S: サーバーの現在時刻で TOTP(K) を計算 S->>S: 入力されたコードと比べる 端末とサーバーの間で通信するのは利用者が入力したコードだけで、端末はネットワークにつながっていなくても、コードを作れる。 TOTP の既定値 TOTP では、コードの有効時間・桁数・HMAC の種類を選べる。既定値は次のとおりである。 | 値 | 既定値 | 意味 | 根拠 | |---|---|---|---| | 有効時間 | 30秒 | 1つのコードが有効な時間の単位。短いほど盗まれたコードの寿命は短くなるが、入力が間に合わなくなる。その釣り合いとして推奨されている | RFC 6238 §5.2 | | 桁数 | 6桁 | 勘で当たる確率は、1回あたり 1/1,000,000 | RFC 4226 §5.3 | | HMAC の種類 | HMAC-SHA-1 | HOTP と同じ。TOTP では HMAC-SHA-256 や HMAC-SHA-512 に変えてもよい | RFC 6238 §1.2 | 本記事では3つとも既定値を使う。既定値から変えない理由は、2.3 の補足で説明する。 2.3 秘密鍵の受け渡し(登録) TOTP ではサーバーと端末が同じ秘密鍵を持っていることが前提になる。では、秘密鍵をどうやって端末に渡すか。 2.3.1 登録は最初の1回だけ 秘密鍵を端末に渡す作業を、登録という。利用者ごとに最初の1回だけ行う。 1. サーバーが、利用者ごとの秘密鍵を作り、保存する 2. 秘密鍵を QR コードにして画面に表示する 3. 利用者が、認証アプリで QR コードを読み取る 4. 認証アプリに表示されたコードを入力してもらい、正しければ登録を確定する 秘密鍵が通信路を通るのは、この1回だけである。以降のログインでは6桁のコードしか送らない。 4 で確認するのは、読み取りに失敗していた場合や別の端末で読み取ってしまった場合に、確定する前に気づけるようにするためである。確認せずに確定すると本人がログインできなくなる。 秘密鍵の長さは最低128ビット、推奨160ビットとされている(RFC 4226 §4)。 2.3.2 otpauth:// URI QR コードの中身は、次の形式の文字列(URI)である。 otpauth://totp/{issuer}:{account}?secret={秘密鍵}&issuer={issuer} | 項目 | 説明 | |---|---| | totp | 種類。時刻ベースなら totp、カウンターベースなら hotp | | {issuer}:{account} | 認証アプリに表示される名前。サービス名とアカウント名 | | secret | 秘密鍵。Base32 で書く | | issuer | サービス名。ラベルの issuer と同じにする | | algorithm・digits・period | 省略時は SHA1・6桁・30秒 | この形式は RFC ではなく、Google Authenticator が公開した仕様(Key Uri Format)である。多くの認証アプリが対応しており、事実上の標準になっている。 QR コードは、この URI をそのまま画像にしたものである。つまり、QR コードの画像は秘密鍵そのものである。よってスクリーンショットを残したり、外部の QR 生成サービスに URI を送ったりしてはいけない。 <details><summary>補足:なぜ既定値から変えないのか</summary> Key Uri Format の仕様ページには、Google Authenticator は algorithm と period を無視する、と書かれている(最終更新は2018年で、現行版での扱いは未確認)。サーバーだけ SHA-256 や60秒にしても、アプリが既定値で計算すればコードは一致しない。このため SHA-1・6桁・30秒の既定値から変えないのが安全である。 </details> これでサーバーと端末が同じ秘密鍵を持ち、同じ計算ができるようになった。ではサーバーはコードを照合するとき、何に気をつけるか。 2.4 サーバー側の照合 2.4.1 時刻のずれを許す サーバーと端末の時刻が少しでもずれているとT が1つ違い、コードが一致しない。また30秒の切り替わりの直前にコードを入力すると、届くまでに次の T に変わってしまう。 そこでサーバーは照合する時点(現在時刻)の T のコードだけでなく、その前後のステップのコードも受け付ける。RFC 6238 は通信の遅れとして多くても1ステップの許容を推奨している(RFC 6238 §5.2)。前後1ステップを許すと「1つ前」「現在」「1つ後」の3つのステップのコードを受け付けることになる。 ただし受け付けるコードが増えると、次の2つの副作用が出る。 | 副作用 | 内容 | 対策 | |---|---|---| | 当たりやすくなる | 1回あたりの当たる確率が、1/1,000,000 から 3/1,000,000 になる | 2.4.2 | | 使い回しやすくなる | 1つのコードが、最大で約90秒間有効になる | 2.4.3 | 2.4.2 試行回数を制限する 6桁は100万通りしかないので、何回でも試せるなら総当たりでいずれ当たる。そこで失敗の回数に上限を設けて上限に達したら一定時間ロックする(RFC 4226 §7.3)。 - 回数はログインのたびではなく利用者ごとに数える。ログインし直すだけで回数が戻ると制限の意味がない - 上限は使い勝手を損なわない範囲でなるべく小さくする。NIST SP 800-63B は、連続失敗の上限を100回までとしている(NIST SP 800-63B-4 §3.2.2) 2.4.3 同じコードを2回使えなくする コードが最大で約90秒有効だと利用者が入力したコードを肩越しに見てすぐに使われるおそれがある(リプレイ攻撃)。RFC 6238 は一度成功したコードの2回目を受け付けてはならない(MUST NOT)としている(RFC 6238 §5.2)。 これを実装するには、利用者ごとに「最後に成功したタイムステップ番号」を保存し、それ以下のステップのコードを拒否する。 2.4.4 秘密鍵を暗号化して保存する サーバーは照合のたびに秘密鍵からコードを計算する。このため元の秘密鍵を持っておく必要がある。 | | パスワード | TOTP の秘密鍵 | |---|---|---| | 照合の方法 | 入力をハッシュ化して、保存した値と比べる | 秘密鍵からコードを計算して、入力と比べる | | 保存の方法 | ハッシュ化(元に戻せない) | 暗号化(照合のときに復号する) | | 漏れたとき | ハッシュから元に戻すのは難しい | 秘密鍵があれば、いつでも正しいコードを作れる | 秘密鍵はハッシュ化できないので暗号化して保存し、照合するときだけ復号する(RFC 6238 §5.1)。暗号化の鍵は、DB と別の場所に置く。 本記事では、暗号化に Python の cryptography ライブラリに入っている Fernet を使う。 <details><summary>用語解説:Fernetとは</summary> Fernet は、Python の cryptography ライブラリに入っている共通鍵暗号の仕組みである(cryptography Fernet)。1つの鍵で、暗号化と復号の両方を行う。 - 暗号化には AES(128ビットの鍵、CBC モード)を使う - 改ざんされていないかを確かめるために、HMAC-SHA256 を付ける - 鍵は、32バイトの値を Base64 で書いた44文字の文字列である - 暗号文は gAAAAA で始まる AES と HMAC を自分で組み合わせると、使い方を間違えやすい。Fernet は、それらを正しく組み合わせた決まった手順として使えるので、暗号の細かい設定を考えずに済む。 </details> 2.5 既存のログインに組み込む TOTP を足すと、ログインでは「パスワード」と「コード」の2つを確かめることになる。2つを1回のリクエストで送らせる方法と、パスワードを確かめてからコードを求める2段階の方法がある。2段階にすると、パスワードが違う人にはコードの入力欄を見せずに済み、TOTP を登録していない利用者は1段目でログインを終えられる。 TOTP を足したあと、利用者がパスワードやコードを送る場面は、次の3つになる。 | 場面 | 利用者が送るもの | TOTP か | 行う回数 | 何を確かめるか | 通ったら | |---|---|---|---|---|---| | ログインの1段目 | ID・パスワード | TOTP ではない | ログインのたび | パスワードを知っているか(知識要素) | TOTP を登録していればコードを求める。登録していなければログイン完了 | | 登録の確定(2.3.1 の手順4) | 認証アプリの6桁のコード | TOTP | 利用者ごとに最初の1回だけ | QR を正しく読み取れたか。つまり、サーバーと認証アプリが同じ秘密鍵を持てたか | TOTP を有効にする。以後のログインで2段目を求めるようになる | | ログインの2段目 | 認証アプリの6桁のコード | TOTP | ログインのたび | 秘密鍵の入った端末を持っているか(所持要素) | ログイン完了 | ログインの「2段階」は、1段目(パスワード)と2段目(TOTP)の組み合わせである。種類の違う2つの要素を1段ずつ確かめることで、多要素認証になる。 認証アプリのコードを入力する場面は、登録の確定とログインの2段目の2つである。どちらも同じ TOTP のコードで、照合のしかた(2.4)も同じである。違うのは、何のために確かめるかと、通ったあとにサーバーが行うことである。登録の確定はログインではないので、ここでコードを入力しても、ログインしたことにはならない。 2段階にする場合、サーバーは「この利用者はパスワードが通った」ことを、2段目のリクエストまで覚えておく必要がある。そこで、1段目でパスワードが通ったら、短い期限(本記事では5分)の証明を発行する。これをTOTP待ちトークンと呼ぶ。2段目でコードも通ったら、ログイン済みであることを示す証明を発行する。これをアクセストークンと呼ぶ。TOTP待ちトークンは、アクセストークンの代わりに使えてはいけない。使えると、コードを入れずにログインできてしまう。 2.6 TOTPで防げること・防げないこと | 攻撃・事故 | TOTP で防げるか | 理由・対策 | |---|---|---| | パスワードの漏えい | 防げる | パスワードだけではログインできない | | コードの盗み見 | 防げる | 使用済みのコードは使えない(2.4.3) | | 総当たり | 防げる | 試行回数の制限(2.4.2) | | フィッシング | 防げない | 偽のログイン画面に入れたパスワードとコードを、攻撃者がその場で本物の画面に入れればログインできる | | サーバーからの秘密鍵の漏えい | 防げない | 秘密鍵があれば、誰でも正しいコードを作れる。暗号化して保存する(2.4.4) | | 認証アプリのクラウド同期 | 弱まる | 秘密鍵が別の端末にも入り、「その端末を持っている」ことの証明にならなくなる | | 端末の紛失・盗難 | 一部防げる | パスワードも要るので、端末だけではログインできない | NIST SP 800-63B は、OTP にはフィッシング耐性が無いと明記している(NIST SP 800-63B-4 §3.2.5)。フィッシングまで防ぐには FIDO2 / WebAuthn(パスキー)のように、アクセス先のサイトを確かめてから認証する方式が要る。 <details><summary>補足:パスワードのみ・TOTP・FIDO2 / WebAuthn の比較</summary> | | パスワードのみ | パスワード+TOTP | FIDO2 / WebAuthn | |---|---|---|---| | 確かめる要素 | 知識 | 知識+所持 | 所持(+生体・暗証番号) | | サーバーに送るもの | パスワード | パスワードとコード | 端末の秘密鍵で作った署名(秘密鍵は送らない) | | サーバーが持つ秘密 | パスワードのハッシュ | パスワードのハッシュと、TOTP の秘密鍵 | 公開鍵だけ | | フィッシング耐性 | 無い | 無い | ある | | 導入のしやすさ | ― | 既存のログインに足しやすい。認証アプリは無料 | 端末やブラウザの対応が要る | </details> TOTP は既存の ID・パスワードのログインに手軽に所持要素を足せる方式である。3章以降では本章で説明した登録(2.3)と照合(2.4)を実装し、2.5 の2段階の形で既存のログインに組み込む。 2.7 その他の用語 | 用語 | 意味 | |---|---| | Unix 時刻 | 1970-01-01 00:00:00 UTC からの経過秒数。タイムゾーンに左右されないので、日本の端末と海外のサーバーでも同じ T になる | | HMAC | 秘密鍵を使って、データを決まった長さの値に変換する計算。鍵を知らないと、同じ結果を作れない | | Base32 | データを英大文字 A〜Z と数字 2〜7 の32種類の文字だけで表す方法(RFC 4648 §6)。紛らわしい 0 と 1 を使わないので、手入力しやすい | | ハッシュ化 | データを、元に戻せない形に変換すること | | Fernet | Python の cryptography に入っている共通鍵暗号の仕組み。1つの鍵で暗号化と復号を行い、改ざんの検知も付いている | | JWT | 署名付きのトークン。ログインに成功したときにサーバーが発行し、以降のリクエストに付けてログイン済みであることを示す | | NTP | ネットワーク経由で、機器の時計を正しい時刻に合わせる仕組み。TOTP は時刻が命なので、サーバーの時計は NTP で合わせておく | 3. TOTP実装 本章では、2章で説明した TOTP の仕組みを使って、ID・パスワードのログインに TOTP の確認を足す小さな API を作る。次の順に進める。 | 節 | 内容 | |---|---| | 3.1 | 作るものの全体像(ファイル構成、ログインの流れ、エンドポイント、テーブル、設計値) | | 3.2 | 事前準備(Python の仮想環境、パッケージ、暗号化の鍵) | | 3.3 | TOTP の登録と照合の処理を書く(totp.py) | | 3.4 | totp.py を使って API を書く(main.py) | 3.1 作るものの全体像 この記事で作るものは、2.5 で説明した2段階のログインである(流れは 3.1.2 の図を参照)。 TOTP を登録していない利用者には、これまでどおりアクセストークンを返す。TOTP の登録は、ログインした本人が行う(3.1.3 の /totp/enroll)ため、最初から全員に TOTP を求めると、まだ登録していない利用者はログインできず、登録もできなくなるからである。また既存のログインに足す場合、全員が一度に登録を終えられるわけではない。 ただし、TOTP を登録していない利用者はパスワードだけでログインできるままである。実際のシステムに組み込む場合は、全員に登録を必須にするか、いつまでに登録してもらうかを決める必要がある(4.2.1)。 3.1.1 ファイル構成 ~/totp-demo ├── requirements.txt # 使うパッケージ(3.2) ├── totp.py # TOTP の登録と照合(3.3) ├── main.py # API(3.4) ├── demo.db # 起動すると作られる SQLite の DB └── qr.svg # 4.1 で保存する QR コード TOTP そのものの処理は totp.py に、API の処理は main.py に分けた。 3.1.2 ログインの流れ sequenceDiagram autonumber participant U as 利用者 participant S as サーバー U->>S: POST /login(ID・パスワード) S->>S: パスワードを確かめる alt TOTP を登録していない利用者 S->>U: アクセストークン(ログイン完了) else TOTP を登録済みの利用者 S->>U: TOTP待ちトークン(5分) U->>S: POST /login/totp(TOTP待ちトークン+6桁のコード) S->>S: コードを照合する S->>U: 正しければアクセストークン(ログイン完了) end TOTP待ちトークンは、「パスワードは通ったが、TOTP はまだ」という状態を表す。アクセストークンとして使えてはいけない(3.4.1)。 3.1.3 エンドポイント一覧 POST /login と POST /login/totp は3.1.2 の図の工程に対応する。残りの3つはログインしたあと(アクセストークンを持っている状態)で使う。 | エンドポイント | 必要なもの | 3.1.2 の図の工程 | 役割 | |---|---|---|---| | POST /login | ID・パスワード | 1〜4 | パスワードを確かめ(2)、<br>TOTP を登録していなければアクセストークン(3)、<br>登録済みなら TOTP待ちトークン(4)を返す | | POST /login/totp | TOTP待ちトークン、コード | 5〜7 | コードを照合し(6)、<br>正しければアクセストークンを返す(7) | | GET /me | アクセストークン | ―(ログイン後) | ログイン中の利用者を返す | | POST /totp/enroll | アクセストークン | ―(ログイン後) | 秘密鍵を作り、QR コードを返す | | POST /totp/activate | アクセストークン、コード | ―(ログイン後) | 認証アプリのコードが正しければ、TOTP を有効にする | 3.1.4 usersテーブル ID・パスワードの列にTOTP 用の5列を足す。 | 列 | 内容 | |---|---| | totp_secret | 秘密鍵。暗号化して保存する | | totp_enabled | 1 なら、ログインで TOTP を求める | | totp_last_used_step | 最後に成功したタイムステップ番号。再利用の防止に使う | | totp_failures | 続けて失敗した回数 | | totp_locked_until | ロックが解ける時刻(Unix 時刻) | 3.1.5 採用する設計値 | 項目 | 値 | なぜ必要か | 根拠 | |---|---|---|---| | 時刻のずれ | 前後1ステップ(±30秒)まで許す | 端末の時計が少しずれていても、ログインできるようにする | RFC 6238 §5.2 | | 再利用の防止 | 最後に成功したステップ以下のコードは拒否する | 盗み見たコードを、有効なうちに使い回されないようにする | RFC 6238 §5.2 | | 試行回数の制限 | 5回続けて失敗したら15分ロックする | 100万通りを総当たりされないようにする | RFC 4226 §7.3 | | 秘密鍵の保存 | 暗号化して DB に保存する。暗号化の鍵は環境変数で渡す | DB から漏れても、すぐには使えないようにする | RFC 6238 §5.1 | | 方式 | SHA1・6桁・30秒 | 認証アプリとの互換性 | RFC 6238 の既定値 | | 秘密鍵の長さ | 160ビット | 推測されないようにする | RFC 4226 §4 | <details><summary>補足:「5回で15分」の根拠</summary> 「5回で15分」は業界の標準値ではなく、本記事で採用した値である。NIST SP 800-63B では連続失敗の上限を100回までとしている(§3.2.2)。この設定だと1日に試せるのは最大480回で前後1ステップを許しても、1日で当たる確率は約0.14%(480 × 3 / 1,000,000)。 </details> 3.2 事前準備(環境構築) 3.2.0 作業ディレクトリとコードの扱い 作業ディレクトリは ~/totp-demo とする。以降のコマンドは、すべてこのディレクトリの中で実行する。 mkdir -p ~/totp-demo cd ~/totp-demo 3.3 と 3.4 のコードはファイルごとに数回に分けて載せている。「totp.py(1/4)」のように番号を付けているので、上から順に1つのファイルにつなげる。つなげた全体は各節の最後に折りたたんで載せている。 3.2.1 uvでPythonの仮想環境を作る FastAPI 0.162 は Python 3.10 以上が必要である。macOS に最初から入っている python3 は 3.9 なので、そのままでは動かない。そこでuv で Python 3.13 の仮想環境を作る。 uv が入っているかを確かめる。入っていなければ、[uv の記事](TODO: リンク)を参照して入れる。 uv --version uv 0.7.8 (0ddcc1905 2025-05-23) 仮想環境を作る。 uv venv --python 3.13 Using CPython 3.13.5 interpreter at: /usr/local/opt/python@3.13/bin/python3.13 Creating virtual environment at: .venv Activate with: source .venv/bin/activate 筆者の Mac には Homebrew で入れた Python 3.13 があったので、uv はそれを使った。Python 3.13 が入っていない場合は、uv が自動でダウンロードしてから仮想環境を作る。 .venv ができる。この記事では activate せず、.venv/bin/python のようにパスを書いて使う。どの Python で動かしているかがコマンドを見れば分かるようにするためである。 3.2.2 パッケージを入れる このデモで使うパッケージを、requirements.txt というファイルに書き出しておく。requirements.txt は、必要なパッケージとそのバージョンを1行に1つずつ並べたテキストファイルである。こうしておくと次の2つの利点がある。 - 1つのコマンドで、必要なパッケージをまとめて入れられる - == でバージョンを決めておくので、読者の環境にも記事と同じバージョンが入る。バージョンが違うと、動きや実行結果が記事と変わることがある ~/totp-demo に、次の内容で requirements.txt を作る。 fastapi==0.162.0 uvicorn==0.38.0 pyjwt==2.10.1 pyotp==2.10.0 qrcode==8.2 cryptography==46.0.3 | パッケージ | 用途 | |---|---| | fastapi、uvicorn | API を作って動かす | | pyjwt | ログインのトークン(JWT)を作る | | pyotp | TOTP の秘密鍵の作成と、コードの計算 | | qrcode | otpauth:// URI を QR コードにする | | cryptography | 秘密鍵を暗号化する(Fernet) | 以下のコマンドで入れる。 uv pip install -r requirements.txt Resolved 20 packages in 898ms Installed 20 packages in 20ms + cryptography==46.0.3 + fastapi==0.162.0 + pyjwt==2.10.1 + pyotp==2.10.0 + qrcode==8.2 + uvicorn==0.38.0 (ほかの依存パッケージは省略) 3.2.3 暗号化の鍵を環境変数に入れる 秘密鍵を DB に保存するときに使う暗号化の鍵を作る。 環境変数は、それを設定したターミナルの中でだけ有効である。このため、次のコマンドは、4.1 でサーバーを起動するときにも使い続けるターミナルで実行する(4.1 では、このターミナルを「ターミナルA」と呼ぶ)。 export TOTP_ENCRYPTION_KEY=$(.venv/bin/python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())") 入ったかを確かめる。値は実行のたびに変わる。 echo $TOTP_ENCRYPTION_KEY 44文字の英数字(末尾は =)が表示されればよい。 <details><summary>補足:なぜ環境変数で渡すのか</summary> 暗号化の鍵がソースや DB と同じ場所にあると、DB と一緒に漏れたときに、暗号化した意味がなくなる。このため、鍵はソースにも DB にも置かず、起動するときに外から渡す。 </details> <details><summary>補足:ターミナルを閉じたとき</summary> 環境変数は、ターミナルを閉じると消える。新しい鍵を作り直すと、前の鍵で暗号化した秘密鍵は復号できない。その場合は、demo.db を消して最初からやり直す。認証アプリに残った登録も消しておく。 </details> 3.3 TOTPの処理を書く(totp.py) 本節では、2章で説明した登録(2.3)と照合(2.4)の処理を、totp.py に書く。totp.py は、3.4 の API から呼ばれる部品であり、次の順に組み立てる。 | 節 | 書くもの | 対応する2章の説明 | |---|---|---| | 3.3.1 | 設計値の定数と、暗号化の準備 | 2.4.4(秘密鍵の暗号化) | | 3.3.2 | 登録:enroll() | 2.3(秘密鍵の受け渡し) | | 3.3.3 | 一致したステップを求める:find_matched_step() | 2.4.1(時刻のずれを許す) | | 3.3.4 | 照合:verify() | 2.4.2(試行回数の制限)、2.4.3(再利用の防止) | ~/totp-demo に totp.py を作り、3.3.1〜3.3.4 のコードを順に書いていく。 3.3.1 定数と暗号化の準備 totp.py の先頭では、このあとの関数が共通で使う値と道具を用意する。3.1.5 で決めた設計値(許容するステップ数、ロックまでの回数と時間)を定数として置き、秘密鍵の暗号化と復号に使う Fernet を、3.2.3 で環境変数に入れた鍵で準備する。 """TOTP の登録と照合を行う部品。 main.py(API)から呼ばれる。このファイルには、次の3つの関数がある。 - enroll() : 登録。秘密鍵を作って DB に保存し、認証アプリに読ませる URI を返す - find_matched_step() : 入力されたコードが、どのタイムステップのコードと一致するかを調べる - verify() : 照合。コードが正しいかを確かめ、再利用の防止と試行回数の制限も行う """ # 環境変数を読むために使う(Python 標準) import os # 現在時刻(Unix 時刻)を得るために使う(Python 標準) import time # TOTP の秘密鍵の作成と、コードの計算を行うライブラリ import pyotp # 秘密鍵の暗号化と復号に使う仕組み(記事の 2.4.4) from cryptography.fernet import Fernet # --------------------------------------------------------------------------- # 設計値(記事の 3.1.5 で決めた値) # --------------------------------------------------------------------------- # 認証アプリに表示されるサービス名。アプリには「TOTP-Demo(alice@example.com)」のように出る ISSUER = "TOTP-Demo" # 時刻のずれを、前後何ステップまで許すか。 # 1 なら「1つ前」「現在」「1つ後」の3つのステップのコードを受け付ける(1ステップ=30秒) VALID_WINDOW = 1 # 何回続けて失敗したらロックするか MAX_FAILURES = 5 # ロックする時間(秒)。15分 × 60秒 = 900秒 LOCK_SECONDS = 15 * 60 # --------------------------------------------------------------------------- # 暗号化の準備 # --------------------------------------------------------------------------- # 秘密鍵を DB に保存するときに使う、暗号化の鍵を読み込む。 # 鍵はソースコードにも DB にも書かず、起動するときに環境変数 TOTP_ENCRYPTION_KEY で渡す。 # 環境変数が無いと、ここで KeyError になり、サーバーは起動しない。 fernet = Fernet(os.environ["TOTP_ENCRYPTION_KEY"]) - 3.1.5 の設計値を定数にしている - fernet は、2.4.4 で説明した Fernet で、秘密鍵の暗号化(3.3.2)と復号(3.3.4)に使う - 鍵は 3.2.3 の環境変数から読む。環境変数が無いと KeyError になって起動しない 3.3.2 登録:秘密鍵を作り、暗号化して保存する 2.3.1 の登録の手順のうち、サーバー側の作業である「秘密鍵を作って保存して認証アプリに渡す otpauth:// URI を作る」部分を enroll() 関数として定義する。3.4.5 の /totp/enroll から呼ばれる。 def enroll(db, user): """登録:秘密鍵を作り、暗号化して DB に保存する。 戻り値は otpauth:// で始まる URI(文字列)。 main.py がこれを QR コードにして返し、利用者が認証アプリで読み取る。 この関数では、まだ TOTP を有効にしない(有効にするのは /totp/activate)。 """ # 利用者ごとの秘密鍵を、ランダムに作る。 # 160ビットの値を Base32(英大文字 A〜Z と数字 2〜7)の32文字で表したもの secret = pyotp.random_base32() # 秘密鍵を暗号化して、DB に保存する。 # - fernet.encrypt() は bytes を受け取って bytes を返すので、 # .encode() で文字列 → bytes に、.decode() で bytes → 文字列に変換している # - totp_last_used_step(最後に使ったステップ)を空に戻すのは、 # 秘密鍵を作り直したときに、前の秘密鍵の記録が残らないようにするため # - SQL の ? には、後ろのタプルの値が順に入る(SQL インジェクションを防ぐ書き方) db.execute( "UPDATE users SET totp_secret = ?, totp_last_used_step = NULL WHERE id = ?", (fernet.encrypt(secret.encode()).decode(), user["id"]), ) # 認証アプリに渡す otpauth:// URI を作って返す。 # name はアカウント名(ここではメールアドレス)、issuer_name はサービス名。 # 暗号化する前の秘密鍵が URI に入るので、この URI(と QR)は人に見せてはいけない return pyotp.TOTP(secret).provisioning_uri(name=user["email"], issuer_name=ISSUER) - pyotp.random_base32() は160ビットの秘密鍵を Base32 の32文字で返す - 保存する前に fernet.encrypt() で暗号化する。DB には暗号文だけが入る - totp_last_used_step を空に戻すのは、秘密鍵を作り直したときに、前の秘密鍵で使ったステップが残らないようにするためである - provisioning_uri() は、2.3.2 の otpauth:// URI を作る。name がアカウント名、issuer_name がサービス名になる この関数では TOTP をまだ有効にしない。有効にするのは認証アプリに出たコードが通ってからである(3.4.5)。 3.3.3 一致したステップ番号を求める 照合の準備として入力されたコードがどのタイムステップのコードと一致するかを調べる関数 find_matched_step() を作る。2.4.1 で説明したとおり現在時刻のステップだけでなく、前後1ステップのコードも受け付ける。一致したステップの番号まで求めるのは、3.3.4 で再利用を防ぐのに使うためである。 def find_matched_step(secret, code, now): """入力されたコードが、どのタイムステップのコードと一致するかを調べる。 引数 secret : 秘密鍵(復号した Base32 の文字列) code : 利用者が入力した6桁のコード(文字列) now : 現在時刻(Unix 時刻。秒) 戻り値 一致したタイムステップ番号(整数)。どれとも一致しなければ None。 pyotp には verify() という照合用の関数もあるが、True / False しか返さず、 どのステップで一致したかが分からない。再利用を防ぐにはステップ番号が要るので、自分で調べる。 """ # この秘密鍵でコードを計算する道具を作る(既定値は SHA-1・6桁・30秒) totp = pyotp.TOTP(secret) # 現在のタイムステップ番号を求める。 # T = floor(Unix 時刻 / 30)。// は「割り算の商の整数部分」を求める演算子 current_step = int(now) // totp.interval # offset を -1, 0, +1 と変えながら、「1つ前」「現在」「1つ後」のステップのコードと比べる for offset in range(-VALID_WINDOW, VALID_WINDOW + 1): # totp.at(now, offset) は、現在から offset ステップずらした時刻のコードを返す。 # 比べるときは == ではなく strings_equal を使う。 # == は先頭から比べて、違う文字が見つかった時点で止まるので、 # かかった時間の差から「何文字目まで合っていたか」を推測される余地がある。 # strings_equal は、一致の度合いにかかわらず同じ時間で比べる if pyotp.utils.strings_equal(code, totp.at(now, offset)): return current_step + offset # どのステップのコードとも一致しなかった return None 前後1ステップを許すので「1つ前」「現在」「1つ後」の3つのステップのコードと比べ、一致したもののタイムステップ番号を返す。 <details><summary>補足:なぜ pyotp の verify() を使わないのか</summary> pyotp には照合のための verify() がある。 pyotp.TOTP(secret).verify(code, valid_window=1) しかしverify() は True / False しか返さず、どのステップで一致したかが分からない。再利用を防ぐ(3.3.4.2)には、一致したステップ番号を保存する必要がある。このためat() で各ステップのコードを作り、自分で比べている。 </details> <details><summary>補足:比べるときに strings_equal を使う理由</summary> == で比べると、先頭から何文字目まで一致したかによって比べる時間がわずかに変わる。この時間の差からコードを推測される余地がある。pyotp.utils.strings_equal() は一致の度合いにかかわらず同じ時間で比べる。 </details> 3.3.4 照合:再利用の防止と試行回数の制限 照合の本体である verify() を作る。3.3.3 で一致したステップを求めたうえで、2.4.2 の試行回数の制限と、2.4.3 の再利用の防止を行い、結果を "ok"・"invalid"・"locked" の3つで返す。3.4.3 の check_code() から呼ばれ、ログイン(3.4.4)と登録の確定(3.4.5)の両方で使う。 def verify(db, user, code): """照合:入力されたコードが正しいかを確かめる。 戻り値は次の3つのどれか。 "ok" : コードが正しく、まだ使われていない "invalid" : コードが違う、または使用済み(再利用) "locked" : 失敗が続いたためロック中 ログイン(/login/totp)と登録の確定(/totp/activate)の両方から呼ばれる。 """ # 現在時刻(Unix 時刻)。この関数の中では、この1つの時刻を使い続ける now = time.time() # --- 1. ロック中なら、コードを見ずに断る --- # totp_locked_until にはロックが解ける時刻が入っている(ロックしていなければ空)。 # その時刻がまだ来ていなければロック中。正しいコードが送られても通さない if user["totp_locked_until"] and user["totp_locked_until"] > now: return "locked" # --- 2. コードが一致するかを調べる --- # DB には暗号化した秘密鍵が入っているので、照合するときだけ復号して元に戻す secret = fernet.decrypt(user["totp_secret"].encode()).decode() step = find_matched_step(secret, code, now) # --- 3. 一致したら、同じコードが2回使われていないかを確かめて記録する --- if step is not None: # 「保存済みのステップより新しいときだけ」ステップ番号を更新する。 # 条件(WHERE の後ろ)を UPDATE 文の中に入れているのがポイント。 # - 初めて使うステップなら、条件に合うので1行更新される → 成功 # - 同じコードの2回目なら、条件に合わないので0行しか更新されない → 失敗 # 「SELECT で読んで比べてから UPDATE する」と2回に分けると、 # 同じコードを同時に2回送られたときに、両方とも「まだ使っていない」と読んで通ってしまう。 # 1つの UPDATE 文にすれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。 # あわせて、成功したので失敗回数(totp_failures)を0に戻す updated = db.execute( "UPDATE users SET totp_last_used_step = ?, totp_failures = 0" " WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?)", (step, user["id"], step), ).rowcount # rowcount は、更新した行の数 if updated == 1: return "ok" # --- 4. ここに来たら失敗。失敗を数え、上限に達したらロックする --- # コードが違った場合と、使用済みのコードだった場合の両方がここに来る。 # 回数は TOTP待ちトークンごとではなく、利用者ごとに DB に数える。 # トークンごとに数えると、ログインし直すだけで回数が0に戻ってしまうため db.execute("UPDATE users SET totp_failures = totp_failures + 1 WHERE id = ?", (user["id"],)) # 増やしたあとの失敗回数を読み直す failures = db.execute("SELECT totp_failures FROM users WHERE id = ?", (user["id"],)).fetchone()[0] if failures >= MAX_FAILURES: # 上限に達したら、ロックが解ける時刻(今から15分後)を記録する。 # 失敗回数は0に戻し、ロックが解けたあと、また5回試せるようにする db.execute( "UPDATE users SET totp_failures = 0, totp_locked_until = ? WHERE id = ?", (now + LOCK_SECONDS, user["id"]), ) return "locked" return "invalid" 照合は3つの段階に分かれる。 3.3.4.1 ロック中なら、照合しない ロック中は、正しいコードでも "locked" を返す。ロックが解ける時刻と今の時刻を比べるだけで、コードは見ない。 3.3.4.2 同じコードを2回使えなくする コードが一致したら、そのステップ番号を totp_last_used_step に保存する。次からは、保存したステップ以下のコードを拒否する。 ポイントは、「保存済みのステップより新しいときだけ更新する」という条件を、UPDATE 文の WHERE に入れていることである。 UPDATE users SET totp_last_used_step = ?, totp_failures = 0 WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?) 同じコードの2回目は、WHERE の条件に合わないので、更新した行が0件になる。rowcount が1のときだけ成功とする。 「SELECT で読んで比べてから、UPDATE で書く」と2回に分けると、同じコードを同時に2回送られたときに、両方が「まだ使っていない」と読んでしまい、両方とも通る。条件を UPDATE 文の中に入れれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。 3.3.4.3 失敗を数え、上限に達したらロックする 失敗したら totp_failures を1増やし、5回に達したら15分後の時刻を totp_locked_until に入れる。ロックしたら回数は0に戻し、解除後にまた5回試せるようにする。 回数は、DB の users テーブルに利用者ごとに数える。TOTP待ちトークンごとに数えると、ログインし直してTOTP待ちトークンを取り直すだけで、回数が0に戻ってしまう。 再利用(3.3.4.2)も、失敗として数える。 <details><summary>補足:なぜ秘密鍵はハッシュ化ではなく暗号化なのか</summary> パスワードは、ハッシュ化(元に戻せない変換)して保存する。照合のときは、入力をハッシュ化して比べれば済むからである。 TOTP の秘密鍵は、ハッシュ化できない。照合のたびに、サーバーが元の秘密鍵でコードを計算する必要があるからである。このため、元に戻せる暗号化で保存し、照合するときだけ復号する(RFC 6238 §5.1)。 </details> <details><summary>totp.py の全体</summary> """TOTP の登録と照合を行う部品。 main.py(API)から呼ばれる。このファイルには、次の3つの関数がある。 - enroll() : 登録。秘密鍵を作って DB に保存し、認証アプリに読ませる URI を返す - find_matched_step() : 入力されたコードが、どのタイムステップのコードと一致するかを調べる - verify() : 照合。コードが正しいかを確かめ、再利用の防止と試行回数の制限も行う """ # 環境変数を読むために使う(Python 標準) import os # 現在時刻(Unix 時刻)を得るために使う(Python 標準) import time # TOTP の秘密鍵の作成と、コードの計算を行うライブラリ import pyotp # 秘密鍵の暗号化と復号に使う仕組み(記事の 2.4.4) from cryptography.fernet import Fernet # --------------------------------------------------------------------------- # 設計値(記事の 3.1.5 で決めた値) # --------------------------------------------------------------------------- # 認証アプリに表示されるサービス名。アプリには「TOTP-Demo(alice@example.com)」のように出る ISSUER = "TOTP-Demo" # 時刻のずれを、前後何ステップまで許すか。 # 1 なら「1つ前」「現在」「1つ後」の3つのステップのコードを受け付ける(1ステップ=30秒) VALID_WINDOW = 1 # 何回続けて失敗したらロックするか MAX_FAILURES = 5 # ロックする時間(秒)。15分 × 60秒 = 900秒 LOCK_SECONDS = 15 * 60 # --------------------------------------------------------------------------- # 暗号化の準備 # --------------------------------------------------------------------------- # 秘密鍵を DB に保存するときに使う、暗号化の鍵を読み込む。 # 鍵はソースコードにも DB にも書かず、起動するときに環境変数 TOTP_ENCRYPTION_KEY で渡す。 # 環境変数が無いと、ここで KeyError になり、サーバーは起動しない。 fernet = Fernet(os.environ["TOTP_ENCRYPTION_KEY"]) def enroll(db, user): """登録:秘密鍵を作り、暗号化して DB に保存する。 戻り値は otpauth:// で始まる URI(文字列)。 main.py がこれを QR コードにして返し、利用者が認証アプリで読み取る。 この関数では、まだ TOTP を有効にしない(有効にするのは /totp/activate)。 """ # 利用者ごとの秘密鍵を、ランダムに作る。 # 160ビットの値を Base32(英大文字 A〜Z と数字 2〜7)の32文字で表したもの secret = pyotp.random_base32() # 秘密鍵を暗号化して、DB に保存する。 # - fernet.encrypt() は bytes を受け取って bytes を返すので、 # .encode() で文字列 → bytes に、.decode() で bytes → 文字列に変換している # - totp_last_used_step(最後に使ったステップ)を空に戻すのは、 # 秘密鍵を作り直したときに、前の秘密鍵の記録が残らないようにするため # - SQL の ? には、後ろのタプルの値が順に入る(SQL インジェクションを防ぐ書き方) db.execute( "UPDATE users SET totp_secret = ?, totp_last_used_step = NULL WHERE id = ?", (fernet.encrypt(secret.encode()).decode(), user["id"]), ) # 認証アプリに渡す otpauth:// URI を作って返す。 # name はアカウント名(ここではメールアドレス)、issuer_name はサービス名。 # 暗号化する前の秘密鍵が URI に入るので、この URI(と QR)は人に見せてはいけない return pyotp.TOTP(secret).provisioning_uri(name=user["email"], issuer_name=ISSUER) def find_matched_step(secret, code, now): """入力されたコードが、どのタイムステップのコードと一致するかを調べる。 引数 secret : 秘密鍵(復号した Base32 の文字列) code : 利用者が入力した6桁のコード(文字列) now : 現在時刻(Unix 時刻。秒) 戻り値 一致したタイムステップ番号(整数)。どれとも一致しなければ None。 pyotp には verify() という照合用の関数もあるが、True / False しか返さず、 どのステップで一致したかが分からない。再利用を防ぐにはステップ番号が要るので、自分で調べる。 """ # この秘密鍵でコードを計算する道具を作る(既定値は SHA-1・6桁・30秒) totp = pyotp.TOTP(secret) # 現在のタイムステップ番号を求める。 # T = floor(Unix 時刻 / 30)。// は「割り算の商の整数部分」を求める演算子 current_step = int(now) // totp.interval # offset を -1, 0, +1 と変えながら、「1つ前」「現在」「1つ後」のステップのコードと比べる for offset in range(-VALID_WINDOW, VALID_WINDOW + 1): # totp.at(now, offset) は、現在から offset ステップずらした時刻のコードを返す。 # 比べるときは == ではなく strings_equal を使う。 # == は先頭から比べて、違う文字が見つかった時点で止まるので、 # かかった時間の差から「何文字目まで合っていたか」を推測される余地がある。 # strings_equal は、一致の度合いにかかわらず同じ時間で比べる if pyotp.utils.strings_equal(code, totp.at(now, offset)): return current_step + offset # どのステップのコードとも一致しなかった return None def verify(db, user, code): """照合:入力されたコードが正しいかを確かめる。 戻り値は次の3つのどれか。 "ok" : コードが正しく、まだ使われていない "invalid" : コードが違う、または使用済み(再利用) "locked" : 失敗が続いたためロック中 ログイン(/login/totp)と登録の確定(/totp/activate)の両方から呼ばれる。 """ # 現在時刻(Unix 時刻)。この関数の中では、この1つの時刻を使い続ける now = time.time() # --- 1. ロック中なら、コードを見ずに断る --- # totp_locked_until にはロックが解ける時刻が入っている(ロックしていなければ空)。 # その時刻がまだ来ていなければロック中。正しいコードが送られても通さない if user["totp_locked_until"] and user["totp_locked_until"] > now: return "locked" # --- 2. コードが一致するかを調べる --- # DB には暗号化した秘密鍵が入っているので、照合するときだけ復号して元に戻す secret = fernet.decrypt(user["totp_secret"].encode()).decode() step = find_matched_step(secret, code, now) # --- 3. 一致したら、同じコードが2回使われていないかを確かめて記録する --- if step is not None: # 「保存済みのステップより新しいときだけ」ステップ番号を更新する。 # 条件(WHERE の後ろ)を UPDATE 文の中に入れているのがポイント。 # - 初めて使うステップなら、条件に合うので1行更新される → 成功 # - 同じコードの2回目なら、条件に合わないので0行しか更新されない → 失敗 # 「SELECT で読んで比べてから UPDATE する」と2回に分けると、 # 同じコードを同時に2回送られたときに、両方とも「まだ使っていない」と読んで通ってしまう。 # 1つの UPDATE 文にすれば、DB が1件ずつ順に処理するので、通るのは片方だけになる。 # あわせて、成功したので失敗回数(totp_failures)を0に戻す updated = db.execute( "UPDATE users SET totp_last_used_step = ?, totp_failures = 0" " WHERE id = ? AND (totp_last_used_step IS NULL OR totp_last_used_step < ?)", (step, user["id"], step), ).rowcount # rowcount は、更新した行の数 if updated == 1: return "ok" # --- 4. ここに来たら失敗。失敗を数え、上限に達したらロックする --- # コードが違った場合と、使用済みのコードだった場合の両方がここに来る。 # 回数は TOTP待ちトークンごとではなく、利用者ごとに DB に数える。 # トークンごとに数えると、ログインし直すだけで回数が0に戻ってしまうため db.execute("UPDATE users SET totp_failures = totp_failures + 1 WHERE id = ?", (user["id"],)) # 増やしたあとの失敗回数を読み直す failures = db.execute("SELECT totp_failures FROM users WHERE id = ?", (user["id"],)).fetchone()[0] if failures >= MAX_FAILURES: # 上限に達したら、ロックが解ける時刻(今から15分後)を記録する。 # 失敗回数は0に戻し、ロックが解けたあと、また5回試せるようにする db.execute( "UPDATE users SET totp_failures = 0, totp_locked_until = ? WHERE id = ?", (now + LOCK_SECONDS, user["id"]), ) return "locked" return "invalid" </details> 3.4 APIを書く(main.py) 本節では3.3 の totp.py を使って、3.1.3 のエンドポイントを持つ API を main.py に書く。次の順に組み立てる。 | 節 | 書くもの | 対応する説明 | |---|---|---| | 3.4.1 | 読み込むライブラリと、トークンに署名する鍵 | 2.5(2種類のトークン) | | 3.4.2 | DB と、動作確認用の利用者 | 3.1.4(users テーブル) | | 3.4.3 | リクエストの形と、エンドポイントで共通に使う関数 | ― | | 3.4.4 | ログイン:/login、/login/totp、/me | 3.1.2(ログインの流れ) | | 3.4.5 | 登録:/totp/enroll、/totp/activate | 2.3(秘密鍵の受け渡し) | ~/totp-demo に main.py を作り、3.4.1〜3.4.5 のコードを順に書いていく。 3.4.1 JWTの鍵を2つに分ける main.py の先頭では、使うライブラリと 3.3 の totp.py を読み込み、ログインで発行する2種類のトークン(2.5 のアクセストークンと TOTP待ちトークン)に署名するための鍵を用意する。 """ID・パスワードのログインに、TOTP の確認を2段目として足した API。 起動すると、次の5つのエンドポイント(URL)が使えるようになる。 - POST /login : ログインの1段目。ID・パスワードを確かめる - POST /login/totp : ログインの2段目。認証アプリのコード(TOTP)を確かめる - GET /me : ログイン中の利用者を返す(ログインできたかの確認用) - POST /totp/enroll : TOTP の登録。秘密鍵を作り、QR コードを返す - POST /totp/activate : 登録の確定。認証アプリのコードが正しければ TOTP を有効にする """ # --- Python 標準のライブラリ --- # 推測されにくいランダムな値を作る。パスワードを安全に比べるのにも使う import secrets # SQLite(ファイル1つで動くデータベース)を使う import sqlite3 # 現在時刻(Unix 時刻)を得る。トークンの有効期限の計算に使う import time # --- 3.2.2 で入れたライブラリ --- # JWT(署名付きのトークン)を作ったり、確かめたりする import jwt # otpauth:// URI を QR コードの画像にする import qrcode import qrcode.image.svg # API を作るためのフレームワーク from fastapi import Depends, FastAPI, HTTPException, Response from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer # リクエストで受け取る JSON の形を定義する from pydantic import BaseModel # --- 3.3 で作った TOTP の部品 --- import totp # --------------------------------------------------------------------------- # トークンに署名する鍵 # --------------------------------------------------------------------------- # ログインで発行する2種類のトークンに、それぞれ別の鍵で署名する。 # - ACCESS_KEY : アクセストークン(ログイン済みの証明)用 # - PENDING_KEY : TOTP待ちトークン(パスワードは通ったが TOTP はまだ、という証明)用 # 同じ鍵にすると、TOTP待ちトークンがアクセストークンの確認を通ってしまい、 # TOTP を入れずに API を使えてしまう。鍵を分ければ、それを防げる。 # 鍵は起動のたびに作り直す(デモなので、再起動したらログインし直せばよい)。 # token_hex(32) は、32バイトのランダムな値を16進数の文字列(64文字)で返す ACCESS_KEY = secrets.token_hex(32) PENDING_KEY = secrets.token_hex(32) アクセストークン用(ACCESS_KEY)とTOTP待ちトークン用(PENDING_KEY)で、署名の鍵を分けている。 TOTP待ちトークンは「パスワードは通ったが、TOTP はまだ」という状態を表す。もしアクセストークンと同じ鍵で署名すると、TOTP待ちトークンも署名の確認を通ってしまい、TOTP を飛ばして API を使えてしまう。鍵を分けておけば、TOTP待ちトークンはアクセストークンの確認(3.4.3)で必ず失敗する。 鍵は起動のたびに secrets.token_hex() で作り直している。デモなので再起動したらログインし直せばよい。 3.4.2 DBと動作確認用の利用者 次に、3.1.4 の users テーブルを SQLite に作り、4.1 の動作確認で使う利用者を1人入れておく。 # --------------------------------------------------------------------------- # データベース(users テーブル) # --------------------------------------------------------------------------- # demo.db というファイルの DB を開く(無ければ作る)。ファイルは、起動したディレクトリにできる。 # - isolation_level=None : SQL を1文実行するたびに、自動で確定(コミット)する # - check_same_thread=False : FastAPI は複数のスレッドから DB を使うので、それを許す db = sqlite3.connect("demo.db", isolation_level=None, check_same_thread=False) # 検索結果を user["email"] のように、列の名前で取り出せるようにする db.row_factory = sqlite3.Row # users テーブルを作る(すでにあれば何もしない)。 # id・email・password が元からあるログインの列、totp_ で始まる5列が TOTP のために足した列 db.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL, password TEXT NOT NULL, totp_secret TEXT, -- 秘密鍵(暗号化して保存) totp_enabled INTEGER NOT NULL DEFAULT 0, -- 1 なら TOTP を求める totp_last_used_step INTEGER, -- 最後に成功したタイムステップ番号 totp_failures INTEGER NOT NULL DEFAULT 0,-- 続けて失敗した回数 totp_locked_until REAL -- ロックが解ける時刻(Unix 時刻) )""") # 動作確認用の利用者 alice を入れる(すでにいれば何もしない)。 # パスワードは、デモのため平文で保存している。実際のシステムではハッシュ化して保存する db.execute("INSERT OR IGNORE INTO users (email, password) VALUES ('alice@example.com', 'password')") # API の本体を作る。このあとの @app.post(...) などで、エンドポイントを足していく app = FastAPI() - 起動すると demo.db ができ、users テーブル(3.1.4)と、動作確認用の利用者 alice が入る - isolation_level=None で、SQL を1文ずつ自動で確定(コミット)させている - パスワードは、デモのため平文で保存している(4.2.1) 3.4.3 リクエストの形と共通の処理 各エンドポイントで共通して使う部品を用意する。前半(3/6)は、リクエストで受け取る JSON の形の定義である。後半(4/6)は、トークンを発行・確認する関数と、3.3.4 の照合結果を HTTP の応答に変える関数である。 # --------------------------------------------------------------------------- # リクエストで受け取る JSON の形 # --------------------------------------------------------------------------- # FastAPI は、届いた JSON がこの形に合っているかを自動で確かめる。 # 項目が足りないなどの場合は、こちらの処理に来る前に 422 エラーを返す class LoginRequest(BaseModel): """/login で受け取る JSON。例:{"email": "alice@example.com", "password": "password"}""" email: str password: str class TotpLoginRequest(BaseModel): """/login/totp で受け取る JSON。/login で受け取った TOTP待ちトークンと、6桁のコード""" totp_token: str code: str class CodeRequest(BaseModel): """/totp/activate で受け取る JSON。認証アプリに表示された6桁のコード""" code: str # --------------------------------------------------------------------------- # エンドポイントで共通して使う関数 # --------------------------------------------------------------------------- def make_token(email, key, minutes): """トークン(JWT)を作る。 中身は「誰か(sub:メールアドレス)」と「いつまで有効か(exp:Unix 時刻)」の2つ。 渡された鍵で署名するので、鍵を知らない人には同じトークンを作れない。 """ return jwt.encode({"sub": email, "exp": int(time.time()) + minutes * 60}, key, algorithm="HS256") def find_user(token, key): """トークンを確かめて、その利用者を DB から取り出す。 渡された鍵で署名を確かめる。署名が合わない(鍵が違う)、期限が切れている、 中身が壊れている場合は、401(認証エラー)を返して処理を止める。 """ try: email = jwt.decode(token, key, algorithms=["HS256"])["sub"] except jwt.PyJWTError: raise HTTPException(401, "Invalid token") return db.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone() def current_user(cred: HTTPAuthorizationCredentials = Depends(HTTPBearer())): """ログイン中の利用者を返す。ログインが必要なエンドポイントで使う。 リクエストのヘッダー「Authorization: Bearer <トークン>」からトークンを取り出し、 アクセストークン用の鍵(ACCESS_KEY)で確かめる。 TOTP待ちトークンは PENDING_KEY で署名しているので、ここでは必ず 401 になる。 """ return find_user(cred.credentials, ACCESS_KEY) def check_code(user, code): """TOTP のコードを照合し、失敗なら HTTP のエラーを返して処理を止める。 totp.verify() の結果を、HTTP の応答に変える。 - "locked" → 429(Too Many Requests:試しすぎ) - "invalid" → 401(認証エラー) - "ok" → 何もせずに戻る(呼び出し元の処理が続く) """ result = totp.verify(db, user, code) if result == "locked": raise HTTPException(429, "Too many failed attempts") if result == "invalid": raise HTTPException(401, "Invalid code") - make_token() は、メールアドレスと有効期限を入れたトークンを、渡された鍵で署名する - find_user() は、渡された鍵で署名を確かめ、利用者を返す。鍵が違えば 401 にする - current_user() は、アクセストークンの鍵(ACCESS_KEY)でしか確かめない。このため、TOTP待ちトークンを送ると 401 になる - check_code() は、3.3.4 の結果を HTTP の応答に変える。ロック中は 429、コードが違えば 401 にする 3.4.4 ログインを2段階にする 3.1.2 の図のログインの流れ(工程1〜7)を、/login(工程1〜4)と /login/totp(工程5〜7)の2つのエンドポイントで実装する。あわせて、ログインできたかを確かめるための /me も作る。 # --------------------------------------------------------------------------- # ログイン # --------------------------------------------------------------------------- @app.post("/login") def login(req: LoginRequest): """ログインの1段目:ID・パスワードを確かめる(3.1.2 の図の工程1〜4)。 - TOTP を登録していない利用者 → アクセストークンを返す(ログイン完了) - TOTP を登録済みの利用者 → TOTP待ちトークン(5分)を返す(2段目へ進む) """ # メールアドレスで利用者を探す user = db.execute("SELECT * FROM users WHERE email = ?", (req.email,)).fetchone() # 利用者がいない、またはパスワードが違えば 401。 # どちらの場合も同じメッセージにして、「そのメールアドレスが登録されているか」を教えないようにする。 # compare_digest は、一致の度合いにかかわらず同じ時間で比べる関数 if user is None or not secrets.compare_digest(req.password, user["password"]): raise HTTPException(401, "Invalid email or password") # TOTP を登録済みなら、アクセストークンの代わりに TOTP待ちトークン(有効期限5分)を返す if user["totp_enabled"]: return {"totp_token": make_token(user["email"], PENDING_KEY, 5)} # TOTP を登録していなければ、これまでどおりアクセストークン(有効期限30分)を返す return {"access_token": make_token(user["email"], ACCESS_KEY, 30)} @app.post("/login/totp") def login_totp(req: TotpLoginRequest): """ログインの2段目:認証アプリのコードを確かめる(3.1.2 の図の工程5〜7)。 TOTP待ちトークンとコードが正しければ、アクセストークンを返す。 """ # TOTP待ちトークンを、TOTP待ちトークン用の鍵(PENDING_KEY)で確かめ、利用者を取り出す user = find_user(req.totp_token, PENDING_KEY) # コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる check_code(user, req.code) # 照合に成功したので、アクセストークン(有効期限30分)を返す return {"access_token": make_token(user["email"], ACCESS_KEY, 30)} @app.get("/me") def me(user=Depends(current_user)): """ログイン中の利用者を返す。アクセストークンが無いと 401 になる。 Depends(current_user) と書くと、この関数が呼ばれる前に current_user() が実行され、 その戻り値(ログイン中の利用者)が user に入る。 """ return {"email": user["email"], "totp_enabled": bool(user["totp_enabled"])} - /login は、パスワードが正しいとき、TOTP が有効ならTOTP待ちトークン(5分)を、無効ならアクセストークンを返す。TOTP を使わない利用者の動きは、今までと変わらない - /login/totp は、TOTP待ちトークンを PENDING_KEY で確かめてから、コードを照合する - /me は、アクセストークンでしか呼べない。4.1 で、TOTP待ちトークンがアクセストークンとして通らないことを確かめるのに使う 3.4.5 登録と有効化 最後に、TOTP を登録するためのエンドポイントを作る。2.3.1 の登録の手順のうち、秘密鍵を作って QR コードを返す部分(手順1〜2)を /totp/enroll が、認証アプリのコードを確かめて登録を確定する部分(手順4)を /totp/activate が受け持つ。 # --------------------------------------------------------------------------- # TOTP の登録 # --------------------------------------------------------------------------- @app.post("/totp/enroll") def enroll(user=Depends(current_user)): """TOTP の登録:秘密鍵を作り、QR コード(SVG 画像)を返す(2.3.1 の手順1〜2)。 ログインした本人が使う(アクセストークンが要る)。 この時点では、まだ TOTP は有効にならない。/totp/activate でコードが通ってから有効になる。 """ # すでに有効なら、登録し直させない(秘密鍵が勝手に作り直されないようにする)。409 は「状態が合わない」 if user["totp_enabled"]: raise HTTPException(409, "Already enabled") # 秘密鍵を作って DB に保存し、otpauth:// URI を受け取る(totp.py の enroll()) uri = totp.enroll(db, user) # URI を QR コードにする。QR はサーバーの中で作る。 # 外部の QR 生成サービスに URI を送ると、秘密鍵が外に漏れるため。 # SvgPathFillImage は、白い背景付きの SVG 画像を作る svg = qrcode.make(uri, image_factory=qrcode.image.svg.SvgPathFillImage).to_string() # SVG 画像として返す。curl の -o で保存し、ブラウザなどで開いて認証アプリで読み取る return Response(svg, media_type="image/svg+xml") @app.post("/totp/activate") def activate(req: CodeRequest, user=Depends(current_user)): """登録の確定:認証アプリのコードが正しければ、TOTP を有効にする(2.3.1 の手順4)。 QR を正しく読み取れたか(サーバーと認証アプリが同じ秘密鍵を持てたか)を確かめるためのもので、 ログインではない。 """ # まだ /totp/enroll で秘密鍵を作っていなければ、確かめようがないので 400(リクエストの誤り) if user["totp_secret"] is None: raise HTTPException(400, "Call /totp/enroll first") # コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる check_code(user, req.code) # 照合に成功したので TOTP を有効にする。以後のログインでは2段目(/login/totp)を求める db.execute("UPDATE users SET totp_enabled = 1 WHERE id = ?", (user["id"],)) return {"totp_enabled": True} - /totp/enroll は、秘密鍵を作り、QR コードを SVG で返す。QR はサーバーの中で作る(外部の QR 生成サービスに送ると、秘密鍵が漏れる) - /totp/activate は、認証アプリに出たコードが通ったら、TOTP を有効にする <details><summary>補足:なぜ確認してから有効にするのか</summary> QR を読み取った時点で有効にすると、読み取りに失敗していた場合や、別の端末で読み取ってしまった場合に、本人がログインできなくなる。認証アプリに出たコードを1回入れてもらい、通ってから有効にすれば、この事故を防げる。 確認に使ったコードは、3.3.4.2 で使用済みになる。このため、確認と同じ30秒の間は、そのコードでログインできない。 </details> <details><summary>main.py の全体</summary> """ID・パスワードのログインに、TOTP の確認を2段目として足した API。 起動すると、次の5つのエンドポイント(URL)が使えるようになる。 - POST /login : ログインの1段目。ID・パスワードを確かめる - POST /login/totp : ログインの2段目。認証アプリのコード(TOTP)を確かめる - GET /me : ログイン中の利用者を返す(ログインできたかの確認用) - POST /totp/enroll : TOTP の登録。秘密鍵を作り、QR コードを返す - POST /totp/activate : 登録の確定。認証アプリのコードが正しければ TOTP を有効にする """ --- Python 標準のライブラリ --- 推測されにくいランダムな値を作る。パスワードを安全に比べるのにも使う import secrets SQLite(ファイル1つで動くデータベース)を使う import sqlite3 現在時刻(Unix 時刻)を得る。トークンの有効期限の計算に使う import time --- 3.2.2 で入れたライブラリ --- JWT(署名付きのトークン)を作ったり、確かめたりする import jwt otpauth:// URI を QR コードの画像にする import qrcode import qrcode.image.svg API を作るためのフレームワーク from fastapi import Depends, FastAPI, HTTPException, Response from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer リクエストで受け取る JSON の形を定義する from pydantic import BaseModel --- 3.3 で作った TOTP の部品 --- import totp --------------------------------------------------------------------------- トークンに署名する鍵 --------------------------------------------------------------------------- ログインで発行する2種類のトークンに、それぞれ別の鍵で署名する。 - ACCESS_KEY : アクセストークン(ログイン済みの証明)用 - PENDING_KEY : TOTP待ちトークン(パスワードは通ったが TOTP はまだ、という証明)用 同じ鍵にすると、TOTP待ちトークンがアクセストークンの確認を通ってしまい、 TOTP を入れずに API を使えてしまう。鍵を分ければ、それを防げる。 鍵は起動のたびに作り直す(デモなので、再起動したらログインし直せばよい)。 token_hex(32) は、32バイトのランダムな値を16進数の文字列(64文字)で返す ACCESS_KEY = secrets.token_hex(32) PENDING_KEY = secrets.token_hex(32) --------------------------------------------------------------------------- データベース(users テーブル) --------------------------------------------------------------------------- demo.db というファイルの DB を開く(無ければ作る)。ファイルは、起動したディレクトリにできる。 - isolation_level=None : SQL を1文実行するたびに、自動で確定(コミット)する - check_same_thread=False : FastAPI は複数のスレッドから DB を使うので、それを許す db = sqlite3.connect("demo.db", isolation_level=None, check_same_thread=False) 検索結果を user["email"] のように、列の名前で取り出せるようにする db.row_factory = sqlite3.Row users テーブルを作る(すでにあれば何もしない)。 id・email・password が元からあるログインの列、totp_ で始まる5列が TOTP のために足した列 db.execute(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY, email TEXT UNIQUE NOT NULL, password TEXT NOT NULL, totp_secret TEXT, -- 秘密鍵(暗号化して保存) totp_enabled INTEGER NOT NULL DEFAULT 0, -- 1 なら TOTP を求める totp_last_used_step INTEGER, -- 最後に成功したタイムステップ番号 totp_failures INTEGER NOT NULL DEFAULT 0,-- 続けて失敗した回数 totp_locked_until REAL -- ロックが解ける時刻(Unix 時刻) )""") 動作確認用の利用者 alice を入れる(すでにいれば何もしない)。 パスワードは、デモのため平文で保存している。実際のシステムではハッシュ化して保存する db.execute("INSERT OR IGNORE INTO users (email, password) VALUES ('alice@example.com', 'password')") API の本体を作る。このあとの @app.post(...) などで、エンドポイントを足していく app = FastAPI() --------------------------------------------------------------------------- リクエストで受け取る JSON の形 --------------------------------------------------------------------------- FastAPI は、届いた JSON がこの形に合っているかを自動で確かめる。 項目が足りないなどの場合は、こちらの処理に来る前に 422 エラーを返す class LoginRequest(BaseModel): """/login で受け取る JSON。例:{"email": "alice@example.com", "password": "password"}""" email: str password: str class TotpLoginRequest(BaseModel): """/login/totp で受け取る JSON。/login で受け取った TOTP待ちトークンと、6桁のコード""" totp_token: str code: str class CodeRequest(BaseModel): """/totp/activate で受け取る JSON。認証アプリに表示された6桁のコード""" code: str --------------------------------------------------------------------------- エンドポイントで共通して使う関数 --------------------------------------------------------------------------- def make_token(email, key, minutes): """トークン(JWT)を作る。 中身は「誰か(sub:メールアドレス)」と「いつまで有効か(exp:Unix 時刻)」の2つ。 渡された鍵で署名するので、鍵を知らない人には同じトークンを作れない。 """ return jwt.encode({"sub": email, "exp": int(time.time()) + minutes * 60}, key, algorithm="HS256") def find_user(token, key): """トークンを確かめて、その利用者を DB から取り出す。 渡された鍵で署名を確かめる。署名が合わない(鍵が違う)、期限が切れている、 中身が壊れている場合は、401(認証エラー)を返して処理を止める。 """ try: email = jwt.decode(token, key, algorithms=["HS256"])["sub"] except jwt.PyJWTError: raise HTTPException(401, "Invalid token") return db.execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone() def current_user(cred: HTTPAuthorizationCredentials = Depends(HTTPBearer())): """ログイン中の利用者を返す。ログインが必要なエンドポイントで使う。 リクエストのヘッダー「Authorization: Bearer <トークン>」からトークンを取り出し、 アクセストークン用の鍵(ACCESS_KEY)で確かめる。 TOTP待ちトークンは PENDING_KEY で署名しているので、ここでは必ず 401 になる。 """ return find_user(cred.credentials, ACCESS_KEY) def check_code(user, code): """TOTP のコードを照合し、失敗なら HTTP のエラーを返して処理を止める。 totp.verify() の結果を、HTTP の応答に変える。 - "locked" → 429(Too Many Requests:試しすぎ) - "invalid" → 401(認証エラー) - "ok" → 何もせずに戻る(呼び出し元の処理が続く) """ result = totp.verify(db, user, code) if result == "locked": raise HTTPException(429, "Too many failed attempts") if result == "invalid": raise HTTPException(401, "Invalid code") --------------------------------------------------------------------------- ログイン --------------------------------------------------------------------------- @app.post("/login") def login(req: LoginRequest): """ログインの1段目:ID・パスワードを確かめる(3.1.2 の図の工程1〜4)。 - TOTP を登録していない利用者 → アクセストークンを返す(ログイン完了) - TOTP を登録済みの利用者 → TOTP待ちトークン(5分)を返す(2段目へ進む) """ # メールアドレスで利用者を探す user = db.execute("SELECT * FROM users WHERE email = ?", (req.email,)).fetchone() # 利用者がいない、またはパスワードが違えば 401。 # どちらの場合も同じメッセージにして、「そのメールアドレスが登録されているか」を教えないようにする。 # compare_digest は、一致の度合いにかかわらず同じ時間で比べる関数 if user is None or not secrets.compare_digest(req.password, user["password"]): raise HTTPException(401, "Invalid email or password") # TOTP を登録済みなら、アクセストークンの代わりに TOTP待ちトークン(有効期限5分)を返す if user["totp_enabled"]: return {"totp_token": make_token(user["email"], PENDING_KEY, 5)} # TOTP を登録していなければ、これまでどおりアクセストークン(有効期限30分)を返す return {"access_token": make_token(user["email"], ACCESS_KEY, 30)} @app.post("/login/totp") def login_totp(req: TotpLoginRequest): """ログインの2段目:認証アプリのコードを確かめる(3.1.2 の図の工程5〜7)。 TOTP待ちトークンとコードが正しければ、アクセストークンを返す。 """ # TOTP待ちトークンを、TOTP待ちトークン用の鍵(PENDING_KEY)で確かめ、利用者を取り出す user = find_user(req.totp_token, PENDING_KEY) # コードを照合する。失敗なら、ここで 401 か 429 が返り、処理が止まる check_code(user, req.code) # 照合に成功したので、アクセストークン(有効期限30分)を返す return {"access_token": make_token(user["email"], ACCESS_KEY, 30)} @app.get("/me") def me(user=Depends(current_user)): """ログイン中の利用者を返す。アクセストークンが無いと 401 になる。 Depends(current_user) と書くと、この関数が呼ばれる前に current_user() が実行され、 その戻り値(ログイン中の利用者)が user に入る。 """ return {"email": user["email"], "totp_enabled": bool(user["totp_enabled"])} --------------------------------------------------------------------------- TOTP の登録 --------------------------------------------------------------------------- @app.post("/totp/enroll") def enroll(user=Depends(current_user)): """TOTP の登録:秘密鍵を作り、QR コード(SVG 画像)を返す(2.3.1 の手順1〜2)。 ログインした本人が使う(アクセストークンが要る)。 この時点では、まだ TOTP は有効にならない。/totp/activate でコードが通ってから有効になる。 """ # すでに有効なら、登録し直させない(秘密鍵が勝手に作り直されないようにする)。409 は「状態が合わない」 if user["totp_enabled"]: raise HTTPException(409, "Already enabled") # 秘密鍵を作って DB に保存し、otpauth:// URI を受け取る(totp.py の enroll()) uri = totp.enroll(db, user) # URI を QR コードにする。QR はサーバーの中で作る。 # 外部の QR 生成サービスに URI を送ると、秘密鍵が外に漏れるため。 # SvgPathFillImage は、白い背景付きの SVG 画像を作る svg = qrcode.make(uri, image_factory=qrcode.image.svg.SvgPathFillImage).to_string() # SVG 画像として返す。curl の -o で保