google

Stars and mentions

Last 30 days, Oct 4 – Nov 2

GitHub stars20,014
mentions387
05601.1K1.7K2.2KOct 4Oct 18Nov 2
UTC days · repositories google owns206 repositories

Package downloads

Cumulative recorded downloads · since 2025-10-17

crates.io—
npm13B
PyPI—
03.3B6.5B9.8B13BOct 4Oct 18Nov 2
Nov 1crates.io: 0/33 packages observed · npm: 3/55 packages observed · PyPI: 0/24 packages observed

Across 112 known packages in repositories google owns. Cumulative totals include recorded downloads before the selected range. Missing observations leave gaps; totals may be incomplete. PyPI excludes known mirrors.

Top repositories

google’s repositories ranked by new stars, as on trending.

Loading

Contributes to

Repositories outside google’s account where google opened or reviewed pull requests in the past 365 days.

Who noticed google

Everyone who starred or forked google’s repositories since the archive began in 2011.

Notable accounts

Ordered by the stars their own repositories have received.

Biggest fans

Accounts that starred the most of these repositories.

Breakout weeks

Each repository’s best week of new stars, with the release published around it.

Mentions

391posts
Nov 38 posts
  1. Hacker News · · cimnine · google/libphonenumber

    Google's common library for parsing, formatting, and validating phone numbers

    1 point
  2. Agent Mesh v0.2.0 Deep in Agent Mesh my agent waited, unheard. Now: one path. Hello. Your agents can now take calls from outside the mesh. Same identity, same policy, one URL each. github.com/google/agentmesh/releases/tag/v0.2.0

    15 likes · 3 reposts · 409 views
  3. Your Agent Changed Last Night and Nobody Deployed Anything: Skill Drift on Google Cloud Skill Registry The incident that never shows up in your deploy log Here is a scenario that's easy to create on Gemini Enterprise Agent Platform today. You build a Managed Agent. You attach a skill from Skill Registry called invoice-triage. It works great. You ship it. Two weeks later, someone on another team updates invoice-triage to handle a new vendor format. They test it with their own agent. It works for them. That same afternoon, your production agent starts misrouting invoices. Nobody touched your agent. No pull request, no deploy, no change ticket. Your CI pipeline is green. Your agent config in Git is identical to last week. What changed is the skill underneath it. I call this skill drift, and in this post I will show you exactly why it happens, how to measure a second hidden problem that comes with it (the context tax of mounting many skills), and a small lockfile pattern that fixes both. Everything here is based on the public Managed Agents API and Skill Registry docs. Both are Preview, so treat the code as a starting point and test it in a sandbox project first. Quick refresher: how skills reach your agent A skill is a folder with a SKILL.md file plus optional scripts/, references/, and assets/. You upload it to Skill Registry as a zipped payload, and the registry keeps a revision history for every update. When you create or update a Managed Agent, you mount skills into the agent's sandbox through base_environment.sources. The docs give you three ways to point at the registry: | Source path | What the agent gets | |---|---| | projects/P/locations/L/skills/SKILL_ID | The default version, whatever it is right now | | projects/P/locations/L/skills/SKILL_ID/skill_versions/VERSION | One exact version, frozen | | projects/P/locations/L/skills | Up to 100 skills from that project and location | The first and third options are the ones people copy from quickstarts. Both are floating references. That is the root of the problem. Problem 1: floating references are :latest tags for agent behavior We learned this lesson years ago with container images. Deploying myapp:latest to production means your production state is defined by whoever pushed last, not by what you reviewed. A skill path without a version is the same thing, except worse in one way: a skill is mostly natural language instructions. A tiny wording change in SKILL.md can change which tool the model calls, what it refuses, or how it formats output. There is no compiler error. There is no failing unit test unless you wrote evals for it. A few details from the docs make this sharper: - UpdateSkill can replace the display name, description, and the whole zipped filesystem in one call. - The skill ID becomes the folder name inside the agent sandbox, so the agent keeps "seeing" the same skill name even after the content changed. - Skill Registry inherits project level IAM. Anyone with write access in the project can update any skill in it. So the blast radius of one skill update is every agent that mounts that skill without a version. Problem 2: the context tax of "mount everything" The third path above is tempting. One line, and your agent has every skill in the project. But think about how skills work. In the Agent Skills pattern, the agent sees each skill's name and description up front so it can decide which one to open. The full SKILL.md body only loads when the skill is picked. That is called progressive disclosure, and it keeps things cheap when you have a handful of skills. With up to 100 skills mounted, those names and descriptions add up. That text rides along on every turn, whether or not any of those skills are relevant. There is also a quality cost. More skills with overlapping descriptions means more chances for the model to pick the wrong one. Two skills named invoice-triage and invoice-review with similar descriptions is a routing bug waiting to happen. You cannot manage what you do not measure, so let's measure it. Measuring the skill metadata budget This script lists every skill in a project and location, joins the names and descriptions the way an agent would roughly see them, and counts tokens with the Count Tokens API. # skill_context_tax.py import os import requests import google.auth from google.auth.transport.requests import Request from google import genai PROJECT = os.environ["GCP_PROJECT_ID"] LOCATION = os.environ.get("SKILL_LOCATION", "us-central1") MODEL = os.environ["MODEL_ID"] # the Gemini model your agent uses BASE = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT}/locations/{LOCATION}" def authed_session(): creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) creds.refresh(Request()) s = requests.Session() s.headers["Authorization"] = f"Bearer {creds.token}" return s def list_skills(s): skills, token = [], None while True: params = {"pageToken": token} if token else {} r = s.get(f"{BASE}/skills", params=params) r.raise_for_status() body = r.json() skills.extend(body.get("skills", [])) token = body.get("nextPageToken") if not token: return skills def main(): s = authed_session() skills = list_skills(s) catalog = "\n".join( f"- {sk.get('displayName', '')}: {sk.get('description', '')}" for sk in skills ) client = genai.Client(vertexai=True, project=PROJECT, location="global") total = client.models.count_tokens(model=MODEL, contents=catalog).total_tokens print(f"Skills in {LOCATION}: {len(skills)}") print(f"Metadata tokens per turn (floor estimate): {total}") if skills: print(f"Average per skill: {total / len(skills):.1f}") if __name__ == "__main__": main() Two honest caveats about this number: 1. It is a floor. The base agent may wrap each skill in extra formatting, file paths, or instructions that you cannot see. 2. It uses the registry description. If the description in your SKILL.md frontmatter is longer, the real cost is higher. Keep the two in sync. To turn it into a monthly number, multiply by your turns per month and your model's input price. Here is the math with made up example numbers, just to show the shape: 80 skills x 70 tokens = 5,600 tokens of skill metadata per turn 5,600 x 2,000,000 turns/month = 11.2 billion input tokens per month Plug in your own skill count, turn volume, and the current price from the pricing page. If the result surprises you, that is the point of measuring. The fix: a skill lockfile The pattern is the same one we use for package managers. Floating references are fine for development. Production gets exact, reviewed versions recorded in a file that lives in Git. The flow looks like this: author updates skill -> registry creates new revision -> staging agent (floating) picks it up -> evals run against staging -> pass? update skills.lock.json in a PR -> merge -> CI pins prod agent to locked versions Step 1: generate and check the lockfile # skill_lock.py import json import os import sys import requests import google.auth from google.auth.transport.requests import Request PROJECT = os.environ["GCP_PROJECT_ID"] LOCATION = os.environ.get("SKILL_LOCATION", "us-central1") BASE = f"https://{LOCATION}-aiplatform.googleapis.com/v1beta1/projects/{PROJECT}/locations/{LOCATION}" LOCKFILE = "skills.lock.json" def authed_session(): creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) creds.refresh(Request()) s = requests.Session() s.headers["Authorization"] = f"Bearer {creds.token}" return s def latest_revision(s, skill_id): revs, token = [], None while True: params = {"pageToken": token} if token else {} r = s.get(f"{BASE}/skills/{skill_id}/revisions", params=params) r.raise_for_status() body = r.json() revs.extend(body.get("skillRevisions", [])) token = body.get("nextPageToken") if not token: break if not revs: return None revs.sort(key=lambda rev: rev["createTime"]) newest = revs[-1] return {"revision": newest["name"].split("/")[-1], "created": newest["createTime"]} def lock(skill_ids): s = authed_session() data = {"project": PROJECT, "location": LOCATION, "skills": {}} for sid in skill_ids: data["skills"][sid] = latest_revision(s, sid) with open(LOCKFILE, "w") as f: json.dump(data, f, indent=2) print(f"Wrote {LOCKFILE} with {len(skill_ids)} skills") def check(): s = authed_session() with open(LOCKFILE) as f: data = json.load(f) drift = [] for sid, locked in data["skills"].items(): current = latest_revision(s, sid) if current is None or current["revision"] != locked["revision"]: drift.append((sid, locked["revision"], current and current["revision"])) for sid, old, new in drift: print(f"DRIFT {sid}: locked {old}, registry now has {new}") sys.exit(1 if drift else 0) if __name__ == "__main__": if sys.argv[1] == "lock": lock(sys.argv[2:]) elif sys.argv[1] == "check": check() Usage: python skill_lock.py lock invoice-triage vendor-lookup refund-policy python skill_lock.py check # exits 1 if any skill moved The check command is your drift alarm. It does not block anyone from updating a skill. It just tells you, loudly, that the registry no longer matches what production was approved with. Step 2: pin the production agent to the lockfile The agent update uses PATCH with update_mask=base_environment. There is no update method in the Python SDK for agents yet, so this goes through REST. One gotcha worth calling out: the docs examples resend the whole base_environment object, including the network allowlist. Treat this update as a full replace, not a merge. If you send only sources, plan for the possibility that other environment settings go back to defaults. Network access is off by default, so a lost allowlist can break an agent that calls external APIs. # pin_agent.py import json import os import requests import google.auth from google.auth.transport.requests import Request PROJECT = os.environ["GCP_PROJECT_ID"] AGENT_ID = os.environ["AGENT_ID"] # The agent docs call this segment "skill_versions", while the registry # returns revision names under "revisions". Confirm which one your project # accepts, then set it here. PIN_SEGMENT = os.environ.get("PIN_SEGMENT", "skill_versions") ALLOWLIST = json.loads(os.environ.get("NETWORK_ALLOWLIST", '[{"domain": "api.example.com"}]')) AGENT_URL = ( f"https://aiplatform.googleapis.com/v1beta1/projects/{PROJECT}" f"/locations/global/agents/{AGENT_ID}?update_mask=base_environment" ) def main(): with open("skills.lock.json") as f: lock = json.load(f) sources = [] for sid, info in lock["skills"].items(): path = ( f"projects/{lock['project']}/locations/{lock['location']}" f"/skills/{sid}/{PIN_SEGMENT}/{info['revision']}" ) sources.append({"type": "skill_registry", "source": path, "target": "/.agent/skills"}) body = { "name": AGENT_ID, "base_environment": { "type": "remote", "sources": sources, "network": {"allowlist": ALLOWLIST}, }, } creds, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"]) creds.refresh(Request()) r = requests.patch( AGENT_URL, headers={"Authorization": f"Bearer {creds.token}", "Content-Type": "application/json"}, json=body, ) r.raise_for_status() print("Update started:", r.json().get("name")) if __name__ == "__main__": main() The update is a long running operation, so poll the returned operation name before you call the rollout done. Step 3: run the drift check on a schedule Any scheduler works. Here is a minimal GitHub Actions version that runs every morning and fails loudly on drift: name: skill-drift-check on: schedule: - cron: "0 12 * * *" workflow_dispatch: jobs: check: runs-on: ubuntu-latest permissions: id-token: write contents: read steps: - uses: actions/checkout@v4 - uses: google-github-actions/auth@v2 with: workload_identity_provider: ${{ secrets.WIF_PROVIDER }} service_account: ${{ secrets.DRIFT_SA }} - run: pip install requests google-auth - run: python skill_lock.py check env: GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }} SKILL_LOCATION: us-central1 The service account only needs read access, so roles/aiplatform.viewer is enough for this job. A few more things I noticed while building this Skill IDs are a namespace you cannot get back quickly. Skill IDs are immutable, and after a delete the ID stays reserved for a while (the docs mention 24 hours in one place and permanent reservation in another, so plan as if it is long). Pick names carefully, and do not use delete plus recreate as a "rollback" strategy. Pinning a previous version is the rollback. Regions do not line up one to one. Managed Agents only support the global location, while Skill Registry lives in specific regions (us-central1, europe-west4, us-east5 at the time of writing). Your mount path includes the skill's region, so write that region into your lockfile instead of assuming it. Cloud Storage mounts skip validation. You can also mount skills straight from a bucket. The docs say the backend does not validate folder content in that case. That makes the registry path the safer choice for production, because uploads go through payload checks and you get revision history. Write descriptions for routing, not marketing. Since descriptions are both your token cost and the model's routing signal, short and specific beats long and friendly. "Classifies AP invoices by vendor and urgency. Not for refunds." does more work than a paragraph. Limitations and caveats - Skill Registry and Managed Agents are Pre-GA. APIs, paths, and field names can change, and the docs say not to use confidential data with the Managed Agents preview. - At the time of writing, Skill Registry does not support VPC Service Controls, CMEK, or HIPAA. Check the compliance table before you design around it. - The docs use two different names for a pinned skill path (skill_versions on the agent side, revisions on the registry side). My script makes this configurable because I could not confirm which is final. Verify in your own project. - The token numbers from the context tax script are a lower bound, not an exact bill. The base agent may add text you cannot see. - A lockfile tells you what changed. It does not tell you whether the change is good. Pair it with evals on a staging agent before promoting. Wrapping up Skills make agents easier to build and share across teams. That same sharing means a skill update is a production change for every agent that mounts it without a version. Three habits fix most of it: pin versions in production, measure what "mount everything" costs you, and run a daily drift check against a lockfile in Git. None of this needs new infrastructure. It is the same discipline we already apply to container images and package managers, pointed at a new kind of artifact. If you try this pattern, I would love to hear what your skill metadata token count came out to. References - Create and manage agents (attach skills, pinning, ListSkills mount): docs.cloud.google.com - Create and manage skills in Skill Registry: docs.cloud.google.com - Google Skills repository (SKILL.md examples): google/skills - Agent Skills format conventions: agentskills.io

    8 reactions
  4. Mozc に 3.35.6354 のタグをつけました 辞書データの更新や変換精度の改善にくわえて、ユーザー辞書の優先順位の見直しや Linux でのダークモードへの対応、コードベースへの広範なリファクタリングなどを含んでいます PRやフィードバックなどご協力ありがとうございました github.com/google/mozc/releases/tag/3.35.6354

    7 likes
  5. #Google published open-source skills for designing and reviewing software with AI agents google/elephant-goldfish-model t.co

    26 likes · 4 reposts · 962 views
  6. # AIに任せるなんてとんでもない!と言われている人達へ — 「AIを信じなくていい仕組み」から始める 「最先端のAI企業のエンジニアは、もうコードを書いていないらしい」「プロンプトを書く時代は終わった」「AIエージェントがひと晩でアプリを作った」——X や YouTube には、こんな話が毎日のように流れてきます。一方、職場では「AIに全部やらせるなんて危ない」という空気がある。どちらを信じればいいのか分からないまま、結局チャット欄にコードを貼っては貼り返す毎日に戻っている。この記事は、そんな人に向けて書きました。 先に結論を書きます。大事なのは、AIをどこまで信じるかではありません。AIを信じなくても困らない仕組みを、先に用意することです。 流れてくる話に、映っていないもの 派手な事例の多くは、たぶん本当です。ただ、記事や動画に映りやすいのは「AIが何をしたか」で、映りにくいのは「それが正しいと、誰がどう確かめたか」です。 自動化がうまくいっているチームは、たいていこの地味な部分を作り込んでいます。 テストが通らなければ、マージできない AIが触れてはいけないファイルは、そもそもAIから見えない AIの「修正しました」という報告は、証拠として扱わない 実際、派手な成果の正体は、この地味な部分だったという報告がいくつも出ています。OpenAI のあるチームは、人間が直接コードを書かずに、約5か月で100万行規模のソフトウェアをAIに作らせたと報告しました。ただ、その報告の大半は「AIに何を書かせたか」ではなく、「AIを迷わせない仕組み」の話です。短く保った指示ファイルや、間違いを指摘するだけでなく直し方まで教えてくれる検査ツールといったものです。LangChain は、AIのモデルを一切変えずに、周りの仕組みだけを作り直して、ベンチマークの順位を上位30位の圏外から上位5位まで上げています(2026年2月の発表時点)。 この部分を飛ばして真似ると、うまくいきません。そして、うまくいかなかった経験が「やっぱりAIに任せるのは危ない」という空気を強めます。派手な話も、慎重な声も、半分しか映していないのです。 人間が毎回見ていても、16回やり直した 私自身の失敗談を書きます。縦書きの本を組むツールを、AIとチャットしながら作っていたときのことです。図がページの境目で切れる、章がページの途中から始まる——こうしたレイアウト崩れを直すのに、20回以上のやり取りが必要になりました。 手を抜いていたわけではありません。毎回スクリーンショットを撮ってAIに見せ、直ったら自分の目で確かめていました。1回の確認は5分ほどです。それでも、20回以上のやり取りのうち16回は、直ったはずのものが直っていなかったためのやり直しでした。「直ったと思ったら、別の場所が崩れていた」「A5では直っていたのに、ほかの用紙サイズでは崩れたままだった」。このツールは用紙4種類×縦書き・横書きの8通りで見た目が変わるので、5分の目視ではとても見切れなかったのです。 つまり、人間がループの中にいても、品質は守られていませんでした。足りなかったのは人間の注意力ではありません。「直ったかどうか」を測る物差しです。 信じるのではなく、信じなくていいようにする そこで、物差しをプログラムで作りました。「どの図もページの境目をまたいでいない」「どの章もページの頭から始まる」を、ブラウザで実際に組んだ結果から測り、だめなら失敗で終わるテストです。8通りすべてに掛けても、人間の手間は増えません。 話はここで終わりません。このテスト自体にも穴がありました。わざと壊したツールに掛けてみたところ、横書きの崩れを一つも検出できていなかったのです。それでもテストは「合格」を返し続けていました。壊した版で試していなければ、何も測っていない物差しを信じていたことになります。 ここから言えるのは、「AIを信じるか、人間が全部見るか」の二択ではない、ということです。 合否を決める物差しを用意する その物差しが本当に測れているかも、わざと壊して確かめる この2つがあれば、AIの出力を一つひとつ信じて確かめる必要はなくなります。物差しが通さなければ、マージされないだけです。ただし物差しも完全ではありません。何を測っていて、何をまだ測っていないかを把握しておくのは、人間の仕事として残ります。 会社で「AIに全部やらせるのは危ない」と言われたら 企業では「中身を理解していないコードを本番に出すのは、説明責任の放棄だ」という声がよく上がります。この心配はもっともです。ただ、少し考えると、私たちはこれまでも、依存しているライブラリのコードを1行ずつ読んできたわけではありません。採用実績、脆弱性の監視、テストや監査の記録といった「信頼してよい根拠」を示すことで、説明責任を果たしてきました。 AIが書いたコードも同じです。人間がすべての行を読むのをやめる代わりに、人間が読むものを変えます。 仕様 —— 何を満たせば完成なのか テスト —— 仕様を正しく写しているか 間違えると被害が大きい箇所の差分 —— 認証、決済、データの削除など 差分をどこまで読むかは、一律に決めなくてかまいません。間違えたときの被害の大きさで変えます。画面の文言の修正ならテストの結果で十分ですし、認証や決済に触れる変更なら、人間が行単位で読みます。 それ以外は、テストの結果と記録で示します。「全部読みました」と言うより、検証できる根拠を示せる分、説明としてはむしろ強くなります。「フル自動は悪」と「全部任せればいい」の間には、この現実的な道があります。 明日からできる、小さな4歩 いきなり全部を自動化する必要はありません。コツは、注意書きを増やすのではなく、修正を「お願い」から「仕組み」へ一段ずつ下ろしていくことです。プロンプトや指示ファイルに書いた注意は、会話が長くなったりモデルが替わったりすると、いつの間にか埋もれます。テストや設定にしたものは、消すまで効き続けます。 4つも多いと感じたら、最初は一つだけでかまいません。次にAIに不具合の修正を頼むとき、直す前に、その不具合を再現するテストを書かせ、それが落ちるところを見せてもらってください。直したあとに同じテストが通れば、「直しました」を言葉ではなく結果で確かめたことになります。 そこから先は、次の4つへ広げていきます。どれも数日で試せて、どれも「AIを信じる」必要がありません。 本物のパスワードや鍵を、AIの作業場所に置かない —— 「.env を読まないで」とお願いするのではなく、そもそもファイルが存在しない状態にします。お願いは破られることがありますが、無いものは読めません。 「直しました」を信じず、CI の結果で判断する —— テストが通らなければマージできないよう、リポジトリの設定(ルールセットやブランチ保護)で縛ります。管理者も含めて例外を許さず、AIの使うアカウントには設定を変える権限を渡さなければ、AIにも人間にも抜け道がなくなります。 AIが一度ハマった失敗は、テストを1本書いて残す —— 注意書きを増やすより確実に、同じ失敗を防げます。そしてそのテストは、わざと壊した版で落ちることを確かめてから使います。 レビューの重心を、コードからテストと仕様へ移す —— テストや仕様の変更には人間の承認が必要になるよう設定し、AIがテストを書き換えて「合格」させる抜け道を塞ぎます。 この4つが揃うと、AIに任せる範囲を少しずつ広げても怖くなくなります。任せる範囲を決めるのは、AIの賢さではなく、物差しの確かさだからです。 もっと知りたい人へ この記事の背景にある考え方は、論文「自律型AIエージェント時代におけるソフトウェア工学の再定義」にまとめました。上の4歩をそのまま始められる雛形(エージェントへの指示ファイル AGENTS.md、仕様のテンプレート、CI とレビューの設定など)と、記事に出てきた組版ツール、そして「物差しの物差し」まで含めたテストも、同じリポジトリで公開しています。 リポジトリ:hamatz/agent-harness-kit 論文:github.com/hamatz/agent-harness-kit/blob/main/doc… リポジトリを取ってきて npm install のあと npm run demo を実行すると、わざと壊した組版ツールが検査で落ち、壊していない版が通る様子を数分で見られます。まずはそこから触ってみてください。 設計の段階については、Google が公開している Elephant-Goldfish Model というプラグインが参考になります。設計書をAIと一緒に書き、それまでの会話を知らない別のAIに査読させる仕組みです。このキットは、その後ろの「実装して、正しいと確かめる」段階を受け持つように作っています。 AIをどこまで信じるかで悩むのは、もうやめていいと思います。信じなくていい仕組みを、一つずつ足していきましょう。

  7. 【CVE-2026-111497】Brotliの脆弱性(Broccoli2Shell) 概要 Googleが開発するデータ圧縮ライブラリ「Brotli」において、細工された圧縮ストリームを展開(デコード)する際に境界外書き込みが発生する脆弱性が公表された。未認証の遠隔の攻撃者から悪意あるBrotli圧縮ストリームを処理させられることで、隣接するデコーダー状態メモリの破壊、機密メモリ内容の漏洩、サービス拒否(DoS)、または任意のコード実行(RCE)に至る可能性がある1。 項目 内容 CVE ID CVE-2026-111497 対象ソフトウェア Google Brotli(C言語デコーダー実装 c/dec/decode.c) CVSSスコア ・9.2 Critical (CVSS:4.0/AV:N/AC:L/AT:P/PR:N/UI:N/VC:H/VI:H/VA:H/SC:N/SI:N/SA:N — Google CNA / GitHub) ・9.8 Critical (CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:U/C:H/I:H/A:H — NVD) ・8.1 High / Important (CVSS:3.1/AV:N/AC:H/PR:N/UI:N/S:U/C:H/I:H/A:H — Microsoft / Red Hat / AWS) 脆弱性種別 CWE-252: 戻り値の未チェック(Unchecked Return Value)、CWE-787: 境界外書き込み(Out-of-bounds Write) 発見・報告者 Khai Tran 氏、Sylvie Mayer 氏(OpenAI Codex Security) 悪用・KEV登録状況 Imperva社がNode.js 24.22.0(Linux arm64公式イメージ)環境でのRCE実証を報告。2026年10月27日付で米国CISA KEVに追加(対策期限: 2026年11月10日) 本脆弱性は、発見から波及範囲の判明に至る経緯の中で以下の3つの通称で呼ばれている。 Node2Shell: 2026年10月21日のNode.jsセキュリティリリース公表後、Imperva Red TeamがNode.js組み込みBrotliデコーダーにおける解析およびRCE実証に命名した呼称である4。 HTTP2Shell: 当初はNode.jsに関連して「Node2Shell」と呼称されていたが、根本原因がHTTP通信(Content-Encoding: br)等で広く利用されるBrotliのC言語実装にあり、複数製品が影響を受けることから「HTTP2Shell」とも呼称されている(HTTP/2プロトコル自体の脆弱性ではない)5。 Broccoli / Broccoli2Shell: セキュリティ研究者のJuliano Rizzo氏が「Node2Shellという名称は影響範囲を狭く見せてしまう」として「Broccoli」と呼ぶことを提唱したことに端を発し、追跡サイト「broccoli2shell.com」等で用いられている呼称である6。 影響を受けるバージョンと修正バージョン 1. Brotli の影響範囲と修正状況 Brotliの公式C実装において、バージョン 1.0.3 から 1.2.0 までのデコーダーが影響を受ける。各系統の修正バージョンは以下の通りである1。 系統 / ブランチ 影響を受けるバージョン 修正版 補足 1.0 系 >= 1.0.3, < 1.0.10 1.0.10 コミット 50a9b70a(0ed4268a のバックポート) 1.1 系 >= 1.1.0, < 1.1.1 1.1.1 コミット da980636(0ed4268a のバックポート) 1.2 系 >= 1.2.0, < 1.2.1 1.2.1 コミット b418fa81 / 0ed4268a 2. 主要ランタイム・言語パッケージ・アプライアンスの対応状況 BrotliのC言語デコーダーをバンドルまたはバインディングとして組み込んでいる各ランタイム・パッケージおよびネットワーク製品の対応状況は以下の通りである。 ランタイム / 製品 影響を受けるバージョン 修正版 補足 Node.js (v24 LTS) 24.x(< 24.22.1) v24.22.1 / v24.22.2 3 v24.22.1 で修正(v24.22.2 はリグレッション修正版) Node.js (v22 LTS) 22.x(< 22.23.4) v22.23.4 組み込みBrotliの更新(2026年10月セキュリティリリース) Node.js (v26 Current) 26.x(< 26.12.1) v26.12.1 組み込みBrotliの更新(EOL系統の <= 20.x, 23.x, 25.x も影響対象) Bun >= 1.0.21, < 1.4.4 1.4.4 8 バンドルするBrotliを 1.2.0 から 1.2.1 へ更新(GHSA-663w-cv8g-jj2p) Python (brotli / PyPI) >= 1.0.3, < 1.2.1 1.2.1 7 PyPI上の公式C拡張パッケージ(GHSA-9v8h-qqgm-4m35) Python (brotlicffi / PyPI) >= 1.0.3, < 1.2.1.0 1.2.1.0 Brotli C実装のCFFIバインディング(GHSA-9v8h-qqgm-4m35) .NET 8 / 9 / 10 8.0.0〜8.0.32 9.0.0〜9.0.21 10.0.0〜10.0.13 Runtime 8.0.33 Runtime 9.0.22 Runtime 10.0.14 9 SDKは 8.0.133/8.0.427、9.0.123/9.0.320、10.0.114/10.0.403 で修正 Palo Alto Networks PAN-OS PAN-OS 12.2 / 12.1 / 11.2 の未修正版 12.2.4-h1 12.1.7-h11 等 6 データプレーンでTLS復号とBrotli検査を有効化している場合のみ影響(デフォルト無効) 3. 主要Linuxディストリビューションの対応状況 OSの共有ライブラリ(libbrotli)およびディストリビューション提供パッケージにおける対応状況は以下の通りである。ディストリビューションによってはNode.jsや.NETパッケージがOS標準の共有ライブラリ(libbrotli)を動的リンクしているため、影響有無がパッケージ単位で異なる点に注意が必要である。 ディストリビューション 対象パッケージ 修正版 補足 Amazon Linux 2023 brotli, libbrotli, python3-brotli nodejs24, dotnet8.0 1.1.0-1.amzn2023.0.2 アドバイザリ: ALAS2023-2026-3334(brotli)、ALAS2023-2026-3330(nodejs24)、ALAS2023-2026-3367(dotnet8.0)。nodejs22・dotnet9.0・dotnet10.0 は影響なし Ubuntu brotli, libbrotli1, python3-brotli dotnet8, dotnet10, dotnet11 24.04 LTS: 1.1.0-2ubuntu0.1 22.04 LTS: 1.0.9-2ubuntu0.1 アドバイザリ: USN-8956-1(Brotli)、USN-8958-1(.NET 8.0.33 / 10.0.14) Debian brotli, libbrotli1 Trixie (13): 1.1.0-2+deb13u1 Bookworm (12): 1.0.9-2+deb12u1 DSA-6582-1 / DLA-4855-1。Debianの nodejs と python-brotlicffi はシステムの brotli を利用 RHEL (Red Hat) brotli, nodejs:24, dotnet8.0, firefox, thunderbird 提供済(各RHSA) RHSB-2026-015(Important)。RHEL 8/9の nodejs:22、RHEL 10の nodejs22/nodejs24、および dotnet9.0/10.0/11.0 はシステムの brotli を利用 4. AWS マネージドサービスの対応状況 AWSの 2026-155-AWS で公表されている主要マネージドサービスの対応状況は以下の通りである10。 対象サービス / リソース 修正版 補足 AWS Lambda(Node.js 22/24/26、.NET 8/9/10) 更新済 マネージドランタイム・ベースイメージ更新済。コンテナイメージ利用時は最新ベースイメージでの再ビルドが必要(Python/Ruby/provided.al2023 はBrotli非同梱) AWS Elastic Beanstalk (Node.js) 6.11.12 Node.js 24.22.1 および 22.23.4 を含むプラットフォームバージョン 6.11.12 を提供済 AWS App Runner / Amplify / CodeBuild / Synthetics 更新済 App Runnerは2026年10月26日に全リージョン更新済(未デプロイのサービスは1回再デプロイが必要) 影響を受ける条件 本脆弱性は、ライブラリがシステム上に存在するだけでは発火せず、「信頼できないBrotli圧縮ストリームを、対象アプリケーションやプロキシがデコード(展開)する」 処理が行われた際に成立する6。 1. 前提となるシステム・ネットワーク構成 攻撃が成立する代表的な経路は以下の4パターンである。 パターン1:クライアントからの圧縮HTTPリクエストボディを展開するサーバー / API クライアントから送信された Content-Encoding: br ヘッダー付きのリクエストボディを、ミドルウェア(Node.jsのExpress 5.0.0〜5.3.0やbody-parser 2.0.0〜2.3.0における raw() / json() 等)やバックエンドアプリケーションが自動または明示的にBrotli展開している構成(なお、素の node:http サーバー自体は単体ではリクエストボディを自動展開しない)。 パターン2:外部サーバーからの圧縮HTTPレスポンスを展開するHTTPクライアント / クローラー 接続先サーバーから返却された Content-Encoding: br レスポンスを自動展開する構成(Node.jsやBunの組み込み fetch()、Brotliパッケージがインストールされた環境でのPython urllib3 / requests / aiohttp / httpx[brotli] 等)。 多くのHTTPクライアントはリクエスト時に Accept-Encoding: br を送信したかどうかにかかわらず、レスポンスヘッダーに Content-Encoding: br が含まれていれば自動的にBrotliデコーダーを呼び出す。そのため、任意のURLを取得するWebhook、リンクプレビュー、クローラー、または非暗号化通信経路上の中間者攻撃において、悪意あるサーバーから一方的に Content-Encoding: br のレスポンスを返されるだけでトリガーされ得る。 パターン3:ファイアウォール等のデータプレーンにおけるトラフィック検査 Palo Alto NetworksのPAN-OS(PA-Series / VM-Series)において、データプレーン上でTLS/SSL復号(Decryption)とBrotliインスペクションの両方が有効化されている場合、通過するBrotliトラフィックをファイアウォール自身が展開するため影響を受ける。 パターン4:デコーダーAPIの直接呼び出し・ファイル処理 アーカイブ、キャッシュ、メッセージキュー等の外部データを、node:zlib(brotliDecompress*、createBrotliDecompress)、Web標準の DecompressionStream、Pythonの brotli.decompress()、.NETの BrotliStream / BrotliDecoder 等で直接展開している場合。 ! 影響を受けないケース: 圧縮(エンコード)のみ・静的 .br ファイル配信のみの構成: 脆弱性はデコーダー(c/dec/decode.c)に存在するため、信頼できるデータをサーバー側でBrotli圧縮して送信するのみの構成や、事前圧縮された静的な .br ファイルをそのまま配信するだけの構成ではデコード処理が行われず、影響を受けない。 .NET のデフォルト挙動: Microsoftのアドバイザリによれば、ASP.NET Coreおよび HttpClient はデフォルトではBrotli圧縮データを自動展開しない。自動展開でBrotliを有効化するよう明示的に設定変更している場合、または System.IO.Compression(BrotliStream / BrotliDecoder)で信頼できないデータを直接展開している場合のみ影響を受ける9。 2. スタンドアロン実行ファイル・コンテナイメージに関する注意点 自己完結型バイナリ(Self-Contained / 単一実行ファイル): .NETのSelf-ContainedデプロイやNativeAOTでビルドされたアプリケーション、およびBunの bun build --compile で生成された実行ファイルは、ビルド時点のランタイム(および脆弱なBrotliデコーダー)をバイナリ内部に埋め込んでいるため、ランタイム更新後にアプリケーションの再ビルドと再デプロイが必要である。 動的リンクとバンドルの違い: Red Hat Enterprise LinuxやDebian、Amazon Linux 2023では、パッケージによってBrotliを内部バンドルしているもの(例:AL2023およびRHEL 8/9の nodejs24、dotnet8.0)と、OSの共有ライブラリ libbrotli を動的リンクしているもの(例:Debianの nodejs、RHEL 8/9およびAL2023の nodejs22、dotnet9.0、dotnet10.0)が混在している。したがって、OSパッケージとランタイムの双方を確認・更新し、実行中のプロセスを再起動する必要がある。 3. 自環境の確認方法 1. Node.js のバージョンおよびバンドルされている Brotli バージョンの確認 node --version node -p "process.versions.brotli" 2. Bun のバージョン確認(1.4.4 以上であるか) bun --version 3. .NET ランタイムのバージョン確認(8.0.33 / 9.0.22 / 10.0.14 以上であるか) dotnet --list-runtimes 4. Python パッケージの確認(Brotli 1.2.1 以上、brotlicffi 1.2.1.0 以上であるか) python3 -m pip show brotli brotlicffi 5. Linux OSパッケージおよびロード済み共有ライブラリの確認 dpkg -l | grep -i brotli # Debian / Ubuntu rpm -qa | grep -i brotli # RHEL / Amazon Linux sudo grep -l brotlidec /proc/[0-9]*/maps 2>/dev/null 脆弱性の技術的な仕組み 前提:Brotliのブロック切り替えと高速パス(Fast Path)/ 安全パス(Safe Path) Brotliの圧縮ストリームは複数のブロックに分割されており、デコーダーはストリームを読み進めながらリテラル(tree_type = 0)、挿入・コピーコマンド(tree_type = 1)、距離コード(tree_type = 2)のブロックタイプを切り替える。 ブロックタイプを切り替える際、デコーダーは c/dec/decode.c 内の DecodeBlockTypeAndLength() を呼び出して次のブロックタイプ番号とブロック長をビットストリームから読み取る。このときBrotliデコーダーは、入力バッファに十分なデータがある場合に境界チェックを省略して高速に処理する Fast Path(safe = 0) と、入力不足を検査しながら処理する Safe Path(safe = 1) の2つの経路を持っている。 原因:戻り値の未チェック(CWE-252)と境界外書き込み(CWE-787) 脆弱性の根本原因は、c/dec/decode.c 内の DecodeBlockTypeAndLength() が返す真偽値 BROTLI_BOOL の仕様と、呼び出し元における戻り値の未チェックにある1。 戻り値 BROTLI_FALSE における2種類の意味の混在: 修正前の DecodeBlockTypeAndLength() は戻り値として単なる真偽値(BROTLI_BOOL)を返していた。この関数は、Safe Pathで入力ビットが一時的に不足している場合(!SafeReadSymbol や !SafeReadBlockLength)に BROTLI_FALSE を返すだけでなく、ストリームのブロックタイプ最大数 max_block_type が 1 以下であるにもかかわらずブロック切り替えが要求されたというストリーム構文エラー(フォーマット不正) の場合(if (max_block_type <= 1) return BROTLI_FALSE;)にも同じ BROTLI_FALSE を返していた。 Fast Path(safe = 0)での戻り値の無視: 修正前のコードでは、Fast Path用のラッパー関数 DecodeLiteralBlockSwitch()、DecodeCommandBlockSwitch()、DecodeDistanceBlockSwitch() の戻り値型が static void と定義されており、内部で呼び出す Decode*BlockSwitchInternal(0, s) の戻り値を完全に無視していた。さらに呼び出し元のマクロ BROTLI_SAFE(METHOD) も、safe == 0 の分岐では戻り値を検証せずにそのまま処理を継続する実装となっていた。 不正な内部状態でのデコード継続と境界外書き込み: 細工された圧縮ストリームによって本来起こり得ないブロック切り替えが要求された際、脆弱なデコーダーは構文エラーで停止せず、「追加入力が必要である」状態と誤認する、あるいはFast Path上でエラーを無視して不整合な内部状態のまま処理を継続してしまう。その結果、内部バッファの終端を越えて攻撃者のバイト列が書き込まれ、隣接するデコーダー状態メモリ(BrotliDecoderState)の破壊や機密メモリの漏洩、さらには任意のコード実行(RCE)へとつながる。 攻撃フロー 修正パッチ Googleの公式リポジトリにおける修正コミット 0ed4268a(および各ブランチへの反映コミット 50a9b70a、da980636、b418fa81)では、以下の変更が行われている2。 c/include/brotli/decode.h に、ブロック切り替えのフォーマット異常を表す新しいエラーコード BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH = -17 を追加。 DecodeBlockTypeAndLength() および各 Decode*BlockSwitch*() 関数の戻り値型を BROTLI_BOOL / void から BrotliDecoderErrorCode に変更し、max_block_type <= 1 のフォーマット異常時は BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH、入力不足時は BROTLI_DECODER_NEEDS_MORE_INPUT、成功時は BROTLI_DECODER_SUCCESS を明示的に返すように修正。 Safe PathとFast Pathの両方でステータスコードを検証し、BROTLI_DECODER_SUCCESS 以外の場合は即座に saveStateAndReturn へジャンプして処理を中断するマクロ BROTLI_SAFE_WITH_STATUS(METHOD) を導入。 c/dec/decode.c // google/brotli commit 0ed4268a より抜粋 -static BROTLI_INLINE BROTLI_BOOL DecodeBlockTypeAndLength( +static BROTLI_INLINE BrotliDecoderErrorCode DecodeBlockTypeAndLength( int safe, BrotliDecoderState* s, int tree_type) { brotli_reg_t max_block_type = s->num_block_types[tree_type]; const HuffmanCode* type_tree = &s->block_type_trees[ @@ -1187,7 +1187,7 @@ brotli_reg_t* ringbuffer = &s->block_type_rb[tree_type * 2]; brotli_reg_t block_type; if (max_block_type <= 1) { - return BROTLI_FALSE; - return BROTLI_DECODER_ERROR_FORMAT_BLOCK_SWITCH; } /* Read 0..15 + 3..39 bits. */ @@ -1197,11 +1197,13 @@ } else { BrotliBitReaderState memento; BrotliBitReaderSaveState(br, &memento); - if (!SafeReadSymbol(type_tree, br, &block_type)) return BROTLI_FALSE; - if (!SafeReadSymbol(type_tree, br, &block_type)) { - return BROTLI_DECODER_NEEDS_MORE_INPUT; - } if (!SafeReadBlockLength(s, &s->block_length[tree_type], len_tree, br)) { s->substate_read_block_length = BROTLI_STATE_READ_BLOCK_LENGTH_NONE; BrotliBitReaderRestoreState(br, &memento); - return BROTLI_FALSE; - return BROTLI_DECODER_NEEDS_MORE_INPUT; } } ... ringbuffer[0] = ringbuffer[1]; ringbuffer[1] = block_type; - return BROTLI_TRUE; - return BROTLI_DECODER_SUCCESS; } -static BROTLI_INLINE BROTLI_BOOL DecodeLiteralBlockSwitchInternal( +static BROTLI_INLINE BrotliDecoderErrorCode DecodeLiteralBlockSwitchInternal( int safe, BrotliDecoderState* s) { - if (!DecodeBlockTypeAndLength(safe, s, 0)) { - return BROTLI_FALSE; - BrotliDecoderErrorCode result = DecodeBlockTypeAndLength(safe, s, 0); - if (result != BROTLI_DECODER_SUCCESS) { - return result; } PrepareLiteralDecoding(s); - return BROTLI_TRUE; - return BROTLI_DECODER_SUCCESS; } -static void BROTLI_NOINLINE DecodeLiteralBlockSwitch(BrotliDecoderState* s) { - DecodeLiteralBlockSwitchInternal(0, s); +static BROTLI_NOINLINE BrotliDecoderErrorCode +DecodeLiteralBlockSwitch(BrotliDecoderState* s) { - return DecodeLiteralBlockSwitchInternal(0, s); } 呼び出し元のマクロも以下のように BROTLI_SAFE_WITH_STATUS へ置き換えられ、safe フラグの有無にかかわらずエラー発生時にデコードが即座に停止するようになった。 c/dec/decode.c +/* NB: METHOD should return BROTLI_DECODER_SUCCESS, BROTLI_DECODER_ERROR_*, or - BROTLI_DECODER_NEEDS_MORE_INPUT; the later two break the processing. */ +#define BROTLI_SAFE_WITH_STATUS(METHOD) \ - { \ - BrotliDecoderErrorCode status; \ - if (safe) { \ - status = Safe##METHOD; \ - } else { \ - status = METHOD; \ - } \ - if (status != BROTLI_DECODER_SUCCESS) { \ - result = status; \ - goto saveStateAndReturn; \ - } \ - } 対策と緩和策 1. 恒久対策:修正バージョンへのアップデートと再ビルド 利用しているOSパッケージ、言語ランタイム、および依存ライブラリを修正バージョンへ更新し、稼働中のプロセスを再起動する。また、単一実行ファイル化(Bunの bun build --compile、.NETのSelf-Contained / NativeAOT)やDockerコンテナイメージを利用している場合は、更新後のランタイム・ベースイメージを用いて再ビルドおよび再デプロイを実施する必要がある。 各環境におけるアップデートコマンド例# 1. Amazon Linux 2023 の場合(ALAS2023-2026-3334 / 3330 / 3367) sudo dnf update brotli nodejs24 dotnet8.0 -y 2. Ubuntu / Debian の場合(USN-8956-1 / USN-8958-1 / DSA-6582-1) sudo apt update && sudo apt install --only-upgrade brotli libbrotli1 python3-brotli -y 3. Node.js の更新(nvm を利用している場合) nvm install 24.22.1 # または 22.23.4 / 26.12.1 4. Python パッケージ(PyPI)の更新(1.2.1 / 1.2.1.0 以上) pip install -U "Brotli>=1.2.1" "brotlicffi>=1.2.1.0" 5. Bun の更新(1.4.4 以上) bun upgrade 2. 一時的な緩和策(アップデートが直ちに実施できない場合) 各ベンダーのアドバイザリおよび技術資料では、即座にパッチを適用できない環境向けに以下の緩和策が案内されている。 WAFやリバースプロキシにおける受信 Content-Encoding: br リクエストの遮断: AWSのセキュリティブレティン 2026-155-AWS 等では、クライアントから送信される Content-Encoding: br ヘッダー付きHTTPリクエストをWAFやプロキシで拒否する緩和策が提示されている。AWSが公開しているAWS WAFカスタムルールの設定例(Countモードで検証後に "Count": {} を "Block": {} に変更する手順)は以下の通りである。なお、AWS WAFは受信リクエストのみを検査するため、アプリケーションが外部サーバーから取得するレスポンスには効かない点に留意が必要である10。 { "Name": "Node2Shell_BROTLI", "Priority": 2, "Statement": { "RegexMatchStatement": { "RegexString": "^\s*br\s*$", "FieldToMatch": { "SingleHeader": { "Name": "content-encoding" } }, "TextTransformations": [ { "Priority": 0, "Type": "LOWERCASE" } ] } }, "Action": { "Count": {} }, "RuleLabels": [ { "Name": "Node2Shell_Brotli" } ], "VisibilityConfig": { "SampledRequestsEnabled": true, "CloudWatchMetricsEnabled": true, "MetricName": "Node2Shell_BROTLI" } } HTTPクライアントおよびPython環境におけるBrotliデコードの抑止: 注意点(Accept-Encoding ヘッダー変更の限界): リクエストヘッダーの Accept-Encoding から br を除外したとしても、悪意あるサーバーはそれを無視して Content-Encoding: br のレスポンスを送りつけることができるため、クライアント側で自動解凍自体が有効なままでは対策として不十分である。 Bun: セキュリティアドバイザリ(GHSA-663w-cv8g-jj2p)では、fetch 呼び出し時に decompress: false を渡すことで、Accept-Encoding: br の送信を抑止するとともにレスポンス自体の自動デコードを無効化する回避策が案内されている8。 Python: 直ちにアップグレードできない場合の緩和策として、Brotli および brotlicffi パッケージ(ならびに httpx[brotli] 等のextra指定)自体をアンインストールし、urllib3、requests、aiohttp 等でBrotliデコーダーが呼び出されない状態(gzip / deflate へのフォールバック)にする方法が挙げられている6。 PAN-OS: ホットフィックス適用までの回避策としてCLIから set system setting brotli-encoding disable を実行(またはWeb UIの Device > Setup > Content-ID から Allow Brotli Encoding のチェックを解除)してBrotliインスペクションを無効化する手順が案内されている。 参考資料・出典一覧 CVE.org - CVE-2026-111497 Google Brotli GitHub Repository - Commit 0ed4268a Node.js - Wednesday, October 21, 2026 Security Releases Imperva Blog - Node2Shell: From Brotli Data to Remote Code Execution (CVE-2026-111497) JPCERT/CC - BrotliのC言語実装における境界外書き込みの脆弱性(CVE-2026-111497)に関する注意喚起 (JPCERT-AT-2026-0034) Broccoli2Shell: Brotli decoder RCE (CVE-2026-111497) in Node.js, Python and more GitHub Advisory Database - GHSA-9v8h-qqgm-4m35 (PyPI brotli / brotlicffi) Bun Security Advisory - GHSA-663w-cv8g-jj2p Microsoft .NET Security Advisory - CVE-2026-111497: Out of bounds write in Brotli leads to RCE AWS Security Bulletin - 2026-155-AWS: CVE-2026-111497: Native Brotli Decoder Remote Code Execution and Memory Disclosure

  8. Reddit · · r/google_antigravity · u/headpiece747 · google/elephant-goldfish-model

    Elephant-Goldfish with Antigravity 2.0 for Improving My Coding Projects I have used spec driven workflows for my coding projects like superpowers, bmad and others that help in antigravity. tried google/elephant-goldfish-model and asked flash3.8 to add it and said it was already in antigravity so i told it to look at my projects and it found a lot to fix with its audit. just another tool to check out to see if it helps. not affiliated not my project or anything.

    7 points · 6 comments
  9. マルチモーダルLLMの個人メモ②Tokenization はじめに この記事はAIで作成されています。multi-modal LLM関連のメモを記載しています。 目次 - Positonal Encoding - [Tokenization(今回)](#tokenization) - [①BPE(Byte Pair Encoding)](#-bpebyte-pair-encoding) - [②Byte-level BPE](#-byte-level-bpe) - [③WordPiece](#-wordpiece) Tokenization LLM は、入力されたテキストをそのまま処理するのではなく、テキストをトークンに分割し、それぞれをトークン ID に変換して処理する。この処理をトークナイズ(Tokenization)と呼ぶ。LLM では、単語をそのままトークンにするのではなく、単語より小さなサブワード単位に分割する方式が広く使われている。 代表的な方式には、次のものがある。 | 方式 | 語彙の構築方法 | 基本的な考え方 | 代表例・関連技術 | | ---------------- | ------------------ | ------------------- | --------------------------- | | ① BPE | トークンをマージして語彙を増やす | 頻出する隣接ペアを統合する | GPT 系の各種トークナイザー | | ② Byte-level BPE | バイトをマージして語彙を増やす | バイト列に BPE を適用する | GPT-2、tiktoken | | ③ WordPiece | スコアに基づいて語彙を構築する | トークンの組み合わせの有用性を評価する | BERT | | ④ Unigram | 大きな候補語彙からトークンを削除する | 確率モデルでテキストを説明する | SentencePiece の Unigram モデル | なお、BPE、WordPiece、Unigram はトークナイズのアルゴリズムであり、SentencePiece や tiktoken はトークナイザーの実装・ライブラリである。これらは同じ階層の概念ではない。 現在の主流 LLM では、BPE 系のトークナイザーが広く使われている。一方、BERT 系では WordPiece、SentencePiece を利用するモデルでは BPE や Unigram などが使われている。トークナイザーの設計は、語彙サイズ、入力系列の長さ、多言語対応、未知の文字列への対応などに影響する。 ① BPE(Byte Pair Encoding) 論文名: Neural Machine Translation of Rare Words with Subword Units URL: arxiv.org 発表日: 2015-08-31 ✅ ポイント - 隣接するトークンのペアを出現頻度に基づいてマージし、新しいトークンを作るアルゴリズム。 - 最初は文字などの小さな単位から始め、頻出ペアを統合することで語彙を増やしていく。 - マージのたびに新しいトークンが語彙に追加される。 - 目標の語彙サイズやマージ回数に達するまで、この処理を繰り返す。 - 学習時に獲得したマージ規則を使い、推論時のテキストをトークンに分割する。 🧪 学習の疑似コード ```python # 1. コーパスを読み込む corpus = load_corpus() # 2. 初期トークン列を作成する(例:文字やバイト単位) tokens = initialize_tokens(corpus) # 3. 初期語彙を作成する vocabulary = initialize_vocabulary(tokens) # 4. マージ規則を初期化する merge_rules = [] # 5. 語彙サイズが目標に達するまで学習する while len(vocabulary) < target_vocab_size: # 隣接するトークンペアの出現頻度を集計 pair_counts = count_adjacent_pairs(tokens) # 最も頻繁に出現するペアを選択 best_pair = argmax(pair_counts) # ペアをマージして新しいトークンを作成 new_token = merge_pair(best_pair) # 語彙に新しいトークンを追加 vocabulary.add(new_token) # マージ規則を追加(学習順が優先順位になる) merge_rules.append(best_pair) # コーパス内の該当するペアを置換 tokens = replace_pair(tokens, best_pair, new_token) # 6. 学習済みの語彙とマージ規則を保存する save_tokenizer(vocabulary, merge_rules) 実際の実装では、単語境界を扱う前処理や、マージの優先順位を管理する仕組みなども必要になる。 🔎 推論時の疑似コード text = input_text() # 学習済みの語彙とマージ規則を読み込む vocabulary, merge_rules = load_tokenizer() # テキストを初期単位に分割 tokens = initialize_tokens(text) # 学習済みのマージ規則を適用 tokens = apply_merge_rules(tokens, merge_rules) # トークンを ID に変換 token_ids = convert_to_ids(tokens, vocabulary) 推論時には、通常、語彙やマージ規則を再学習しない。学習済みの規則を使ってトークン化する。 🧮 BPE の処理イメージ コーパスに次の単語があるとする。 low lower lowest 最初は文字単位に分割する。 l o w l o w e r l o w e s t 例えば、l と o が頻出ペアであれば、これらをマージする。 lo w lo w e r lo w e s t 次に lo と w をマージする。 low low e r low e s t さらにマージを繰り返すことで、lower や lowest などが1つのトークンになる場合もある。 このように、頻出する短い単位を段階的にまとめて、より長いトークンを作る。 ❓ よくある質問 Q. なぜ頻出ペアをマージするの? - 頻繁に現れる文字列を1つのトークンにまとめることで、トークン数を減らしやすくするため。 - 一般的な単語やよく使われる文字列を効率的に表現できるようにするため。 Q. BPE は単語単位で処理するの? - 実装によって前処理は異なるが、一般的な BPE では、単語などの単位に分けてから、その内部をサブワードに分割することがある。 - Byte-level BPE では、バイト列を基本単位として扱う。 Q. 語彙サイズはどのように決まるの? - 初期語彙に、マージで追加されるトークンを加えていく。 - 目標の語彙サイズに達するまでマージを繰り返す。 - 実際の語彙サイズは、モデルの設計によって決まる。 Q. BPE は未知語に対応できるの? - 文字などの初期単位に分解できれば、未知の単語でも既存の単位の組み合わせとして表現できる。 - ただし、初期単位にない文字の扱いは実装に依存する。この問題に対応しやすい方式が Byte-level BPE である。 --- ② Byte-level BPE 代表例: GPT-2、tiktoken 系のトークナイザー GPT-2 論文: Language Models are Unsupervised Multitask Learners URL: cdn.openai.com 発表日: 2019-02-14 ✅ ポイント - 通常の BPE が文字などを初期単位として扱うのに対し、Byte-level BPE はバイトを初期単位として扱う。 - UTF-8 などのバイト列に変換してから BPE を適用する。 - 基本のバイト値は 256 種類であり、すべての Unicode 文字を初期語彙に登録する必要がない。 - バイト列に変換できるため、多言語のテキストや未知の文字列を扱いやすい。 - GPT-2 が採用した代表的なトークナイズ方式である。 🧪 学習の疑似コード # テキストを読み込む corpus = load_corpus() # テキストをバイト列に変換 byte_sequences = encode_utf8(corpus) # 256 種類のバイト値を初期語彙とする vocabulary = initialize_byte_vocabulary() # あとはBPEと同様 ※ 実際の実装では、正規表現などによる前処理や、特殊トークンの管理なども行う。 🔎 推論時の疑似コード text = input_text() # 学習済みのバイト語彙とマージ規則を読み込む vocabulary, merge_rules = load_tokenizer() # UTF-8 のバイト列に変換 byte_sequence = encode_utf8(text) # あとはBPEと同様 🧮 バイト列への変換例 ASCII 文字の a は UTF-8 では1バイトである。 a → 61 一方、日本語の あ は UTF-8 では3バイトである。 あ → E3 81 82 ※ 数値は16進数表記。 Byte-level BPE では、このようなバイト列を基本単位として扱う。 例えば、日本語の文字が初期語彙に登録されていなくても、バイト列に変換できれば、そのバイトの組み合わせとして表現できる。 ただし、実際のトークナイザーでは、マージ規則や前処理によって最終的なトークンの分割結果が異なる。 ❓ よくある質問 Q. なぜバイト単位にするの? - Unicode の文字をすべて初期語彙に用意する必要がないため。 - 256 種類のバイト値を基本単位にできるため。 - 初期語彙にない文字列でも、バイト列に変換して表現しやすいため。 Q. どんな文字列でも処理できるの? - UTF-8 などでバイト列に変換できる文字列に幅広く対応できる。 - ただし、入力の正規化や特殊トークンの扱いなどは別途考慮する必要がある。 - また、処理できることと、少ないトークン数で効率よく表現できることは別である。 Q. 通常の BPE との違いは? - 通常の BPE は文字などを初期単位にする場合がある。 - Byte-level BPE はバイトを初期単位にする。 - そのため、Byte-level BPE は未知の文字列に対応しやすい。 Q. GPT-2 と tiktoken は同じもの? - GPT-2 は言語モデルであり、Byte-level BPE を採用している。 - tiktoken は高速な BPE 系トークナイザーを提供するライブラリである。 - どちらもバイトレベルのトークナイズに関連するが、語彙やマージ規則が同じとは限らない。 --- ③ WordPiece 代表例: BERT 論文名: BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding URL: arxiv.org 発表日: 2018-10-11 ✅ ポイント - BPE と同様に、単語をサブワードに分割するための方式。 - BPE が基本的に隣接ペアの出現頻度を基準にマージするのに対し、WordPiece はペアの有用性を評価するスコアを使う。 - スコアの考え方の一例として、ペアの出現頻度を個々のトークンの出現頻度で正規化する方法がある。 - BERT 系のトークナイザーで使われる。 - ##ing のように、単語の途中に現れるサブワードを ## 付きで表現することがある。 🧪 学習の疑似コード # コーパスを読み込む corpus = load_corpus() # 初期語彙を作成 vocabulary = initialize_vocabulary(corpus) while len(vocabulary) < target_vocab_size: # 隣接ペアの統計を集計 pair_counts = count_adjacent_pairs(corpus) # ペアのスコアを計算 scores = calculate_wordpiece_scores( pair_counts, vocabulary ) # スコアに基づいて候補を選択 best_pair = select_best_pair(scores) # ペアをマージして新しいトークンを追加 vocabulary.add(merge_pair(best_pair)) # コーパス内の該当するペアを置換 corpus = replace_pair(corpus, best_pair) save_tokenizer(vocabulary) これは、スコアに基づいて語彙を構築する考え方を示した疑似コードである。実際の WordPiece の学習アルゴリズムやスコアの定義は、実装によって異なる。 🔎 推論時の疑似コード text = input_text() # 学習済みの語彙を読み込む vocabulary = load_vocabulary() # テキストを単語などの単位に分割 words = preprocess(text) token_ids = [] for word in words: # 語彙を使って単語をサブワードに分割 subwords = wordpiece_tokenize(word, vocabulary) # サブワードを ID に変換 token_ids.extend(convert_to_ids(subwords, vocabulary)) 一般的な WordPiece の推論処理では、学習済みの語彙を使い、単語を表現できるサブワード列に分割する。 BERT 系の実装では、未知語を [UNK] として扱う場合もある。 🧮 WordPiece のスコア ペアの有用性を説明するための近似的なスコアとして、次の式を考える。 $$ \operatorname{score}(x,y)= \frac{f(xy)}{f(x)f(y)} $$ ここで、 - (f(x)):トークン (x) の出現頻度 - (f(y)):トークン (y) の出現頻度 - (f(xy)):(x) と (y) が隣接して現れる頻度 このスコアでは、単にペアが頻繁に出現するかだけでなく、個々のトークンの出現頻度に対して、ペアがどれくらい強く結びついているかを評価できる。 例えば、次の2つのペアがあるとする。 | ペア | ペアの出現頻度 | 個々のトークンの頻度 | スコア | | --------- | ------: | ---------: | ------: | | a + b | 100 | 1000、1000 | 0.0001 | | c + d | 30 | 40、40 | 0.01875 | この例では、a + b のほうが頻出するが、正規化したスコアでは c + d のほうが高くなる。 つまり、単純な出現頻度ではなく、個々のトークンの頻度に対する相対的な結びつきを評価するという考え方である。 なお、この式は WordPiece の考え方を説明するための近似例であり、すべての実装で使用される厳密なスコア式ではない。 🧮 ##ing の意味 WordPiece では、単語の先頭以外に現れるサブワードを ## 付きで表現することがある。 例えば、 playing ↓ play + ##ing この場合、 - play:単語の先頭に現れるサブワード - ##ing:単語の途中に続くサブワード を表す。 ## はトークンの表記上の記法であり、元のテキストに含まれる文字列ではない。 ❓ よくある質問 Q. BPE との違いは? - BPE は基本的に、隣接ペアの出現頻度に基づいてマージする。 - WordPiece は、ペアの有用性を表すスコアに基づいて語彙を構築する。 - ただし、両者ともサブワードを利用し、語彙を構築する方式である。 Q. なぜ ## を付けるの? - 単語の先頭に現れるトークンと、単語の途中に現れるトークンを区別するため。 - 例えば play と ##ing を区別することで、単語の内部構造を表現できる。 Q. WordPiece は未知語に対応できるの? - 単語を既知のサブワードに分割できれば、未知の単語にも対応できる。 - ただし、分割できない場合は [UNK] などで扱う実装もある。 Q. WordPiece は BERT 専用なの? - WordPiece はトークナイズの方式であり、BERT 以外でも利用できる。 - ただし、BERT が代表的な採用例としてよく知られている。 --- ④ Unigram Language Model 代表例: SentencePiece の Unigram モデル 関連 URL: google/sentencepiece ✅ ポイント - BPE のようにペアを繰り返しマージするのではなく、大きな候補語彙から不要なトークンを削除していく方式。 - テキストの分割方法を最初から1つに固定せず、複数の分割候補に確率を割り当てる。 - 各トークンの確率を学習し、コーパス全体をうまく説明できる語彙を構築する。 - トークンを削除した場合に、コーパスの説明能力がどの程度低下するかを評価する。 - 確率推定には EM アルゴリズムなどが利用される。 🧪 学習の疑似コード # コーパスを読み込む corpus = load_corpus() # 大きめの候補語彙を作成 vocabulary = initialize_large_vocabulary(corpus) # 初期確率を設定 probabilities = initialize_probabilities(vocabulary) while len(vocabulary) > target_vocab_size: # 現在の語彙で分割候補の確率を評価 segmentation_probs = evaluate_segmentations( corpus, vocabulary, probabilities ) # EM などを使ってトークンの確率を学習 probabilities = estimate_probabilities( segmentation_probs ) # 各トークンを削除した場合の影響を評価 removal_scores = evaluate_token_removal( corpus, vocabulary, probabilities ) # 説明能力への影響が小さいトークンを削除 candidates = select_removable_tokens(removal_scores) vocabulary = remove_tokens(vocabulary, candidates) # 残った語彙で確率を再学習 probabilities = reestimate_probabilities( corpus, vocabulary ) save_tokenizer(vocabulary, probabilities) 実際のアルゴリズムでは、削除候補の選択数や確率の再推定方法などが定められている。このコードは、処理の流れを示す概念的な疑似コードである。 🔎 推論時の疑似コード text = input_text() # 学習済みの語彙と確率を読み込む vocabulary, probabilities = load_tokenizer() # 入力テキストに対する分割候補を生成 candidates = generate_segmentations(text, vocabulary) # 各分割候補の確率を計算 scores = calculate_segmentation_scores( candidates, probabilities ) # 最も確率の高い分割を選択 best_segmentation = select_best_segmentation(scores) # トークン ID に変換 token_ids = convert_to_ids( best_segmentation, vocabulary ) 推論時には、学習済みの語彙と確率を使って、入力テキストを適切なトークン列に分割する。 実際の実装では、動的計画法などを利用して最適な分割を効率的に求める。 🧮 Unigram の確率モデル テキストの分割候補を (S) とする。 各トークン (t) の確率を (p(t)) とすると、分割候補の確率は次のように表せる。 $$ P(S)=\prod_{t\in S}p(t) $$ 例えば、unhappiness に対して、次の分割候補があるとする。 候補 A: un + happiness 候補 B: un + happi + ness それぞれの分割候補の確率は、 $$ P(A)=p(\text{un})p(\text{happiness}) $$ $$ P(B)=p(\text{un})p(\text{happi})p(\text{ness}) $$ となる。 Unigram では、このような複数の分割候補を確率モデルによって評価する。 実際の学習では、コーパス中の文字列に対して可能な分割を考慮し、全体の尤度が高くなるように語彙と確率を調整する。 🧮 EM アルゴリズムとの関係 Unigram では、EM アルゴリズムなどを使ってトークンの確率を学習する。 - E ステップ:現在のモデルに基づいて、各分割候補の確率や期待出現回数を評価する。 - M ステップ:その結果を使って、各トークンの確率を更新する。 この処理を繰り返すことで、コーパスを説明する確率モデルを学習する。 ただし、EM アルゴリズム自体が、尤度の低いトークンを直接削除するわけではない。 確率モデルの学習と、語彙からトークンを削除する処理は別のステップである。 🧮 語彙削減の考え方 Unigram では、あるトークンを削除した場合に、コーパスの説明能力がどの程度低下するかを評価する。 例えば、 候補語彙: ["un", "happiness", "happi", "ness", "happy"] この中から happi を削除したとしても、別のトークンの組み合わせで同じ文字列を十分に表現できるなら、説明能力の低下は小さい可能性がある。 一方、削除によって多くの文字列をうまく分割できなくなる場合は、説明能力が大きく低下する可能性がある。 このように、トークン単体の出現頻度だけでなく、削除によるコーパス全体への影響を考慮することが重要である。 ❓ よくある質問 Q. BPE との違いは? - BPE は頻出ペアをマージして語彙を増やす。 - Unigram は大きな候補語彙からトークンを削除して語彙を小さくする。 - Unigram は、複数の分割候補を確率的に評価する。 Q. なぜ最初に大きな語彙を用意するの? - 多くの分割候補を最初から用意しておくことで、コーパスを説明するための選択肢を確保するため。 - その後、不要な候補を削除することで、効率的な語彙を構築する。 Q. EM アルゴリズムは何をしているの? - 主に、分割候補を考慮しながらトークンの確率を学習する。 - 語彙の削減は、トークンを削除した場合の説明能力の変化を評価して行う。 Q. Unigram は分割方法をランダムに決めるの? - 確率モデルでは複数の分割候補を扱う。 - 推論時には、最も確率の高い分割を選ぶ方式や、確率に基づいて分割をサンプリングする方式などがある。 - したがって、学習時に複数の候補を扱うことと、推論時に必ずランダムに分割することは別である。

  10. 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 で保

Trending awards, every completed day and week
Gold
1741st place finishes
Silver
1552nd place finishes
Bronze
1513rd place finishes
Awarded repositories
78Owned by google now

Repositories

Ordered by gold, then silver, then bronze.

  1. it-cert-automation-practiceBest: 1st by PR authors, 997 PR authors766566
  2. material-design-iconsBest: 1st by stars, 6,313 stars1012
  3. zxBest: 1st by stars, 5,273 stars831
  4. material-design-liteBest: 1st by stars, 9,272 stars532
  5. A2ABest: 1st by stars, 7,572 stars520
  6. dopamineBest: 1st by stars, 3,566 stars520
  7. web-starter-kitBest: 1st by stars, 3,203 stars513
  8. gumbo-parserBest: 1st by stars, 1,782 stars502
  9. guetzliBest: 1st by stars, 4,660 stars432
  10. A2UIBest: 1st by stars, 2,320 stars410
  11. deepdreamBest: 1st by stars, 2,754 stars376
  12. comprehensive-rustBest: 1st by stars, 3,123 stars353
  13. ioschedBest: 1st by stars, 1,091 stars342
  14. python-fireBest: 1st by stars, 910 stars321
  15. tinkBest: 1st by stars, 2,439 stars311
  16. grumpyBest: 1st by stars, 4,993 stars301
  17. lmctfyBest: 1st by stars, 1,422 stars210
  18. tsunami-security-scannerBest: 1st by stars, 1,094 stars210
  19. flexbox-layoutBest: 1st by stars, 634 stars201
  20. go-cloudBest: 1st by stars, 1,192 stars200
  21. pyringeBest: 1st by stars, 434 stars200
  22. eng-practicesBest: 1st by stars, 2,128 stars163
  23. fontsBest: 1st by stars, 427 stars131
  24. langextractBest: 1st by stars, 808 stars126
  25. lisp-koansBest: 1st by stars, 273 stars121
  26. eleventy-high-performance-blogBest: 1st by stars, 1,895 stars120
  27. flatbuffersBest: 1st by stars, 524 stars113
  28. ageraBest: 1st by stars, 1,094 stars112
  29. magikaBest: 1st by stars, 1,402 stars111
  30. end-to-endBest: 1st by stars, 434 stars110
  31. tracing-frameworkBest: 1st by stars, 326 stars110
  32. physical-webBest: 1st by stars, 497 stars104
  33. filamentBest: 1st by stars, 1,010 stars102
  34. gvisorBest: 1st by stars, 1,388 stars102
  35. formaBest: 1st by stars, 554 stars101
  36. robotoBest: 1st by stars, 759 stars101
  37. seesawBest: 1st by stars, 955 stars101
  38. wuffsBest: 1st by stars, 679 stars101
  39. makaniBest: 1st by stars, 535 stars100
  40. yapfBest: 1st by stars, 468 stars100
  41. adk-pythonBest: 2nd by stars, 1,287 stars031
  42. brotliBest: 2nd by stars, 743 stars022
  43. EarlGreyBest: 2nd by stars, 728 stars021
  44. gemma_pytorchBest: 2nd by stars, 1,000 stars021
  45. gxuiBest: 2nd by stars, 1,905 stars021
  46. robotstxtBest: 2nd by stars, 1,094 stars021
  47. lyraBest: 2nd by stars, 589 stars020
  48. tamperchromeBest: 2nd by stars, 430 stars012
  49. botteryBest: 2nd by stars, 1,310 stars011
  50. git-appraiseBest: 2nd by stars, 797 stars011
  51. lovefieldBest: 2nd by stars, 594 stars011
  52. osv-scannerBest: 2nd by stars, 359 stars011
  53. cdc-file-transferBest: 2nd by stars, 1,204 stars010
  54. data-transfer-projectBest: 2nd by stars, 459 stars010
  55. fully-homomorphic-encryptionBest: 2nd by stars, 945 stars010
  56. google-api-nodejs-clientBest: 2nd by stars, 240 stars010
  57. incremental-domBest: 2nd by stars, 533 stars010
  58. jsactionBest: 2nd by stars, 281 stars010
  59. kubeflowBest: 2nd by stars, 707 stars010
  60. liquidfunBest: 2nd by stars, 493 stars010
  61. nogotofailBest: 2nd by stars, 499 stars010
  62. runeBest: 2nd by stars, 678 stars010
  63. traxBest: 2nd by stars, 880 stars010
  64. WebFundamentalsBest: 2nd by stars, 251 stars010
  65. wireitBest: 2nd by stars, 930 stars010
  66. gemma.cppBest: 3rd by stars, 849 stars004
  67. adk-goBest: 3rd by stars, 301 stars001
  68. axBest: 3rd by stars, 1,365 stars001
  69. binnaviBest: 3rd by stars, 414 stars001
  70. clusterfuzzBest: 3rd by stars, 887 stars001
  71. eddystoneBest: 3rd by stars, 352 stars001
  72. guavaBest: 3rd by stars, 162 stars001
  73. ionBest: 3rd by stars, 685 stars001
  74. mcpBest: 3rd by stars, 354 stars001
  75. OpenSKBest: 3rd by stars, 431 stars001
  76. rejoinerBest: 3rd by stars, 351 stars001
  77. wwwbasicBest: 3rd by stars, 177 stars001
  78. wycheproofBest: 3rd by stars, 477 stars001

A repository earns an award for finishing first, second or third in a completed trending day or week, ranked by new stars, new forks or pull request authors. Awards follow a repository through renames and transfers and count toward its current owner. Weeks run Monday to Sunday, UTC.