Sign In Once, Verify Anywhereでは、基礎を扱いました。Cognitoが何を発行するのか、サービスがトークンを単独でどのように検証できるのか、そして元のStrobeアプリ(A1)がそれをどのように使っているのかです。本稿はその続きです。同じIDトークンを、Strobe Assistant(A2)の中でブラウザからエージェントへ、さらにMCPサーバーへ、そしてA1 APIへと戻るまで追いかけ、すべての経路でトークンが意味を持ち続けるようにしているいくつかの設計判断を見ていきます。
A2がA1から引き継ぐもの
A2はA1のユーザープールとアプリクライアントをそのまま再利用しています。新しいアイデンティティの仕組みは作っていません。A1から次の3つをそのまま引き継いでいます:
- サインインはA1の認証Lambdaを経由します。
POST /v1/auth/loginは{ user, token }を返し、このtokenがCognitoのIDトークンです。 - A1のAPI GatewayのJWTオーソライザーが、引き続きA1の保護されたすべてのルートを守ります。
- どこでも
subがユーザーのアイデンティティです。 所有者チェック(post.userId == sub)と、アップロードのストレージキーのプレフィックス{sub}/{postId}/{fileId}は、これをもとにしています。
A2で新しくなったのは、そのトークンがどこまで移動するかです。ブラウザ → エージェント → MCPサーバー → A1 APIへ戻る、という流れです。
3つの検証者、1つのトークン
同じトークンが3か所でチェックされます:
| 場所 | ライブラリ | タイミング | |
|---|---|---|---|
| V1 | A1のAPI Gateway | マネージドのJWTオーソライザー | A1の保護されたすべてのルート。MCPサーバーからのget_post / get_commentsの呼び出しも含む |
| V2 | エージェント | aws-jwt-verify(auth.ts) | ACPのauthenticateで一度だけ。expはsession/newとsession/promptのたびに再チェック |
| V3 | MCPサーバー | FastMCPのJWTVerifier | /mcpへのすべてのHTTPリクエスト |
ALBは認証を一切行いません。TLSを終端し、/acp*と/mcp*を振り分け、トークンには手を付けずにそのまま渡します。
機能ごとに、認証がどこに現れ、なぜそのチェックが必要なのかをまとめると次のようになります:
| 機能 / フロー | そこで認証が行うこと | 必要な理由 |
|---|---|---|
| WebUIのサインイン | ブラウザがA1のPOST /v1/auth/loginを呼び出し、IDトークンを受け取る | 提供されたACPクライアントには、アイデンティティがまったくありませんでした。それがなければ「自分の写真」という言葉は意味を持ちません |
ACPのハンドシェイク(initialize → authenticate) | エージェントがcognito-id-tokenを提示し、クライアントが_meta.idTokenでトークンを送り、V2が検証する | エージェントは、ツールを開いたりメモリを読んだりする前に、誰にサービスを提供しているのかを知る必要があります |
session/new、session/prompt | 未サインインまたは期限切れなら、requireIdentity()がauthRequiredで拒否する | WebSocketはトークンより長く生き続けることがあるので、有効期限のチェックで古くなった接続がそのまま続くのを防ぎます |
| 接続ごとに1つのエージェント | identityは、その接続のエージェントインスタンス内の変数 | 同じプロセスを共有するユーザー間で、アイデンティティが漏れることはありません |
| MCPの接続 | エージェントのMCPクライアントがAuthorization: Bearer <user's ID token>を送り、V3が検証する | MCPサーバーはALB経由で単独でもアクセスできるため、呼び出し元を無条件に信頼することはできません |
search_my_content、find_similar_to_image | user_id = TokenClaim("sub")がS3 Vectorsのフィルターになる | これが実際のデータ分離です。これがなければ、Aliceの有効なトークンで全員のベクトルを検索できてしまいます |
get_image | 分類行のuserIdがsubと一致しなければならず、一致しなければ「Unknown image.」 | imageIdはモデルが指定する引数なので、保存済みのデータに照らして所有者を改めてチェックする必要があります |
get_post、get_comments | MCPが同じトークンをA1 APIへ転送する(再びV1) | DynamoDBへの2つ目の、より弱い経路を開く代わりに、A1の認可ルールを再利用します |
エージェントのメモリ(get_recent_runs、save_run_summary) | 実行記録はuserId = subで書き込まれ、クエリされる | あるユーザーのチャットのメモリが、別のユーザーのプロンプトに現れてはいけません |
| 定期実行の振り返り | 振り返りLambdaが登録済みユーザーとしてサインインし、同じACP → MCPの経路を通る | 認証の経路が1つなので、無人のジョブにもuserIdフィルターがそのまま適用されます |
| 非同期の取り込み | トークンはなく、所有者はS3キーまたはSQSメッセージから読み取る | トークンはアップロード時にすでにチェック済み(V1)で、そこから先はアイデンティティが信頼できるデータとして運ばれます |
/healthz | 意図的に認証なし | ALBのヘルスチェックはトークンを提示できず、このルートは機密情報を何も返しません |
1回のチャットターンをステップごとに
1. サインイン(ブラウザ → A1 API)。 acp_client/web/main.ts:
async function signIn(email: string, password: string): Promise<string> {
const response = await fetch(`${STROBE_API_BASE_URL}/v1/auth/login`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ email, password }),
});
...
return body.token; // the Cognito ID token
}
WebUIはトークンをlocalStorageではなくsessionStorageに保存するので、タブを閉じると消去されます。いずれにせよ、トークンは1時間で期限切れになります。
2. エージェントが、どのように認証してほしいかを伝える(initialize)。 agent/src/agent.ts:
authMethods: [
{
id: AUTH_METHOD_ID, // "cognito-id-token"
name: "Strobe sign-in",
description: "Pass the Cognito ID token from POST /v1/auth/login as _meta.idToken.",
},
],3. クライアントが自分が誰なのかを証明する(authenticate)。 クライアント側(main.ts):
await nextConnection.agent.request(acp.methods.agent.authenticate, {
methodId: ACP_AUTH_METHOD_ID,
_meta: { idToken: token },
});エージェント側(agent.ts)では、ここでV2が行われます:
.onRequest(acp.methods.agent.authenticate, async (context) => {
const { methodId, _meta } = context.params;
if (methodId !== AUTH_METHOD_ID) throw acp.RequestError.invalidParams(...);
const idToken = (_meta as { idToken?: unknown } | null | undefined)?.idToken;
if (typeof idToken !== "string" || !idToken) throw acp.RequestError.invalidParams(...);
try {
identity = await deps.verifier.verify(idToken);
} catch (error) {
throw acp.RequestError.authRequired(undefined, "Invalid or expired sign-in token.");
}
return {};
})検証器(agent/src/auth.ts)はaws-jwt-verifyを薄くラップしたもので、前回の記事で説明したチェックを実行します:
this.verifier = CognitoJwtVerifier.create({ userPoolId, clientId, tokenUse: "id" });
async verify(idToken: string): Promise<Identity> {
const payload = await this.verifier.verify(idToken);
...
return { sub: payload.sub, displayName, idToken, expiresAt: payload.exp };
}
4. 以降のすべてのリクエストで有効期限を再チェックする。 agent.ts:
const requireIdentity = (): Identity => {
if (!identity) throw acp.RequestError.authRequired(undefined, "Sign in first (authenticate with the Cognito ID token).");
if (isExpired(identity)) {
identity = null;
throw acp.RequestError.authRequired(undefined, "Your sign-in has expired; sign in again.");
}
return identity;
};5. エージェントは、自分のトークンではなくユーザーのトークンでMCPを呼び出す。 agent/src/mcp.ts:
const transport = new StreamableHTTPClientTransport(new URL(this.mcpUrl), {
requestInit: { headers: { Authorization: `Bearer ${this.idToken}` } },
});また、エージェントはsession/newでクライアントが送ってくる**mcpServersのリストを無視します**。ブラウザは信頼できません。もしブラウザがツールサーバーを選べるなら、エージェントを、そしてそれと一緒にユーザーのトークンを、自分が管理するサーバーへ向けさせることができてしまいます。
6. MCPサーバーは、どのツールを実行する前にも再び検証する(V3)。 FastMCPのJWTVerifierには、ユーザープールのJWKSのURL、発行者、アプリクライアントIDを設定しています(設定の全体はパート3にあります)。トークンがない、または不正な場合は、ツールのコードが実行される前に、FastMCPのミドルウェアがHTTP 401を返します。これはデモで見せているステップの1つです。
7. 検証済みのsubがデータのフィルターになる。 これはモデルから渡されるのではなく、サーバーがツールに注入します:
@mcp.tool(annotations={"readOnlyHint": True})
async def search_my_content(
query: str,
media_type: Literal["all", "image", "text"] = "all",
limit: int = 10,
user_id: str = TokenClaim("sub"), # from the verified token, not in the schema
token: AccessToken = CurrentAccessToken(), # raw token, forwarded to A1
) -> dict:その後のすべてのベクトルクエリにはfilter={"userId": user_id}が付きます。
8. A1への呼び出しにも同じトークンを付ける(再びV1)。
async def _api_get(path: str, bearer: str) -> dict:
"""Call the A1 API. The caller's own token is forwarded so A1's authorisation
rules apply unchanged; ..."""
...
response = await client.get(path, headers={"Authorization": f"Bearer {bearer}"})最も重要だった設計判断
userIdは決してツールの引数にしない
A2のセキュリティ上の決定を1つだけ選ぶなら、これです。ツールの結果は信頼できないテキストです。get_commentsは、ほかのユーザーがあなたの投稿に書いたコメントを返します。そしてモデル(特に小さいモデル)は、そこで見つけた指示に従ってしまうことがあります。もしツールのスキーマにuser_idフィールドがあれば、注入されたコメントによって、モデルがそこに他人のIDを入れるよう誘導される可能性があります。それでもすべてのチェックは通ります。トークンは有効で、ツールも普通に動きます。それでもデータは漏れてしまいます。これは、信頼されたコンポーネントが、信頼できない相手の指示に従って動いてしまう問題です。
TokenClaim("sub")を使えば、このパラメーターはtools/listが返すスキーマに含まれないので、モデルには埋めるべき枠そのものがありません。インジェクションにできることは、せいぜい何を検索するかを変えることだけで、それもユーザー自身のデータの中に限られます。デモで私が述べたとおりです:「モデルがインジェクションを読んでしまうのは想定内です。設計は、モデルがそれに抵抗することに依存していません。」
get_imageも同じ原則で説明できます。こちらではimage_idがモデルの指定する引数そのものなので、サーバーは呼び出し元が言ったことではなく、取り込み時に書き込まれた分類行に照らして所有者をチェックします:
row = await asyncio.to_thread(
lambda: classifications_table.get_item(Key={"imageId": image_id}).get("Item")
)
if row is None or row.get("userId") != user_id:
raise ToolError("Unknown image.")「Unknown image.」は「あなたのものではない」場合と「存在しない」場合で同じ答えなので、このエラーからどの画像が存在するのかはわかりません。
トークンはACPのauthenticateの中で運ぶ
この経路では、通常のAuthorization: Bearerヘッダーは使えません。ブラウザのWebSocket APIはリクエストヘッダーを設定できないからです。私は4つの選択肢を比較しました:
| 選択肢 | 評価 |
|---|---|
ACPのauthenticateで_meta.idTokenを使う ✅ 採用 | プロトコルに沿ったやり方で、トークンがURLに現れることもありません。クライアントの小さな変更と、エージェントのハンドラーが必要でした |
| Cookie | WebSocketのハンドシェイクでブラウザが自動的に送ってくれますが、エージェントのドメイン向けに発行する必要があります。また、自動送信されることで、クロスサイトWebSocketハイジャック(CSRFの親戚)が問題になります |
クエリ文字列 wss://…/acp?id_token=… | 簡単ですが、トークンがALBのアクセスログとブラウザの履歴に残り、expまではそこから使えてしまいます |
Sec-WebSocket-Protocolに紛れ込ませる | 動きますが、正当性を説明しなければならないハックです |
ALBに組み込まれているauthenticate-cognitoアクションはどうでしょうか。これはブラウザのセッション向けに作られています。未認証のユーザーをCognitoのホストされたサインインページへリダイレクトし、独自のセッションCookieを保持します。これはMCPサーバーへのBearerでの呼び出しにも、振り返りLambdaにも合いません。さらに、検証をサービスからロードバランサーへ移すことになりますが、そもそもの狙いは各サービスが自分でチェックすることでした。
MCPサーバーでもう一度検証する
エージェントがすでにトークンを検証しているのに、なぜMCPサーバーでもチェックするのでしょうか。これはパート3で詳しく扱ったので、ここでは要点だけを挙げます:
- MCPサーバーはそれ自体が1つの信頼境界です。 URLは公開されています(エージェントでさえALB経由でアクセスします)。そのため「エージェントが送ってくるものは何でも信頼する」は、実際には「URLを見つけた人なら誰でも信頼する」という意味になります。しかも、入り口はALBだけではありません。FargateのタスクはパブリックサブネットでパブリックIPを持って動いており、科目で用意されたセキュリティグループは、MCPサーバーが待ち受けるポート8080を全世界に開いています。タスクのIPを知った人は誰でもALBを完全に迂回でき、そのとき行く手を阻むのはトークンのチェックだけです。
- 再利用できる状態を保てます。 ほかのエージェントやデスクトップのMCPクライアントが接続することもありえますが、サーバーは呼び出し元について何も知らなくても安全なままです。
- 多層防御です。 MCPサーバーは、ユーザーが誰なのかについてエージェントの言い分を鵜呑みにしません。署名付きトークンから自分で
subを取り出します。たとえエージェントにバグがあったり、エージェントが侵害されたりしても、別人だと主張するだけでは済みません。
コストは、リクエストごとにキャッシュ済みの鍵でローカルの署名チェックが1回増えることです。モデルのレイテンシに比べれば、1ミリ秒にもはるかに満たない時間です。
誰も操作していないときのアイデンティティ
A2には人が介在せずに動く部分が2つあり、それぞれアイデンティティの扱い方が異なります。
定期実行の振り返り:借りたアイデンティティ。 Lambdaは登録済みユーザーとしてサインインし、その後はブラウザと同じWebSocketとauthenticateの呼び出しを使います。振り返りLambdaでは:
def _login(email: str, password: str) -> tuple[str, dict]:
body = _api("POST", "/v1/auth/login", body={"email": email, "password": password})
return body["token"], body["user"]
# run_agent(): initialize -> authenticate -> session/new (trigger=schedule) -> session/prompt
client.request("authenticate", {"methodId": AUTH_METHOD_ID, "_meta": {"idToken": token}})
session = client.request("session/new", {"cwd": "/", "mcpServers": [], "_meta": {"trigger": "schedule"}})Bedrockを直接呼び出すのではなくエージェントを経由することで、認証の経路が1つに保たれます。MCPのuserIdフィルターはそのまま適用され、エージェントのメモリにもその実行が記録され、MomentはA1を通してそのユーザーとして書き込まれます(V1)。
正直に言うと限界もあります。Cognitoは、ユーザー本人の認証情報がなければそのユーザーのトークンを発行しません。登録済みアカウントのログイン情報をSecrets Managerに置いているのはそのためです。マシン間通信用のクライアントクレデンシャルのトークンも存在しますが、それはsubが人ではなくアプリクライアントになるアクセストークンなので、ユーザーごとのフィルターには使えません。本来の解決策は、信頼されたサービスがユーザーに代わってトークンを取得できる、Cognitoのカスタム認証フロー(Lambdaトリガー)でしょう。
非同期の取り込み:データとしてのアイデンティティ。 ワーカーはトークンをまったく持っていません。ワーカーはS3キーから読み取った所有者を信頼します。クライアントがアップロードできるのは、V1を通過したあとにアップロードLambdaが署名付きURLを発行した、まさにそのキーだけだからです:
# SQS worker Lambda
# Upload Lambda's convention: keys are always "{userId}/{postId}/{fileId}"
parts = object_key.split("/")
...
user_id, post_id, file_id = parts
このuser_idはフィルター可能なベクトルのメタデータとして保存され、のちにV3のsubと照合されます。テキストのジョブも同じ仕組みで、投稿Lambdaが検証済みのuserIdをSQSメッセージに入れます。
何が秘密で、何が秘密でないか
agent/src/config.tsのコメントが、これをうまく要約しています:
エージェントは秘密情報を一切持たない。AWSの鍵は持たず(タスクロール)、モデルのAPIキーも持たず(BedrockはIAMで認可される)、WebUIと共有するトークンも持たない。接続が続く間、ユーザーのCognitoのアイデンティティを借りるだけだ。
- ユーザープールIDとアプリクライアントID → SSM Parameter Store(標準階層、無料)。 これらは公開の識別子です。すべてのトークンの
issとaudに現れ、トークンを署名するためではなく検証するために必要なものです。 - 振り返り用アカウントのログイン情報 → Secrets Manager。 これは本物の認証情報で、A2が追加した唯一の秘密情報です。
- A2には署名鍵がどこにもありません。 秘密鍵を持っているのはCognitoだけなので、A2のどのサービスも、盗む価値のあるものを何も持っていません。
まとめ
振り返ってみると、A2の認証設計はいくつかの考え方に集約されます:
- 1つのアイデンティティを、すべての境界でチェックする。 A2は新しいアイデンティティの仕組みを追加していません。A1のIDトークンが最後まで移動し、単独でアクセスできる各サービス(A1のAPI、エージェント、MCPサーバー)がそれぞれ自分で検証します。前段のALBは振り分けるだけです。
- アイデンティティは検証済みのトークンから得るものであり、誰かの言い分から得るものでは決してない。 ブラウザからでも、エージェントからでもなく、そして最も重要なことに、モデルからでもありません。
user_id引数を公開する代わりにsubを注入することで、プロンプトインジェクションはデータ漏洩から、最悪でも自分のコンテンツに対する奇妙な検索に変わります。 - トークンがないときは、信頼がどこから来るのかを明示する。 取り込みは、検証済みのアップロードからしか生まれえないストレージキーを信頼します。振り返りは登録済みユーザーのログインを借り、その後はほかの全員と同じ経路を通ります。
次に取り組みたい粗い部分もまだ残っています。WebUIはまだトークンを更新しないので、1時間たつと再びサインインを求められます。また、パート3で触れたとおり、3つのサービスすべてが1つのタスクロールを共有しているため、エージェントとMCPサーバーの間の境界はIAMではなくコードで強制されています。
私にとって最も役に立ったのは、このプロジェクトを通じて身についた習慣です。すべての経路で、*このコンポーネントは、ユーザーが誰なのかをどうやって知っているのか?*と問うことです。その答えが「誰かがそう言ったから」であれば、そこが最初に見るべき場所です。