夏までRoblox（署名検証）

いいテーマきたね。
「署名付きリクエスト」をFlask / FastAPI 両方で“ミドルウェアとして完全にカプセル化”しておこう。

---

フレームワーク別ざっくり比較

項目	Flask	FastAPI
ミドルウェア化	before_request / デコレータ	依存関数 / ミドルウェア
型安全	弱い	強い
導入のしやすさ	既存プロジェクトに楽に追加	新規設計に向く

---

共通：署名検証ロジック（HMAC-SHA256）

import hmac
import hashlib

SECRET = b"YOUR_SECRET_KEY"
HEADER_NAME = "X-Signature"

def verify_signature(raw_body: bytes, signature: str | None) -> bool:
    if not signature:
        return False
    expected = hmac.new(SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)

---

Flask版：完全ミドルウェア化

1. Blueprint 用の署名ミドルウェア

from flask import Blueprint, request, abort, jsonify

api = Blueprint("api", __name__)

@api.before_request
def check_signature():
    # この Blueprint 配下の全ルートに適用
    raw_body = request.get_data(cache=False)
    signature = request.headers.get(HEADER_NAME)

    if not verify_signature(raw_body, signature):
        return jsonify({"error": "unauthorized"}), 401

2. ルート定義

@api.post("/ai/generate")
def ai_generate():
    data = request.get_json()
    # ここに来た時点で署名は検証済み
    return jsonify({"ok": True, "data": data})

3. アプリへの登録

from flask import Flask
from api import api

app = Flask(__name__)
app.register_blueprint(api, url_prefix="/api")

---

FastAPI版：依存関数としてのミドルウェア化

1. 署名検証依存関数

from fastapi import Depends, Header, HTTPException, status, Request

async def verify_signature_dep(
    request: Request,
    x_signature: str | None = Header(default=None, alias=HEADER_NAME),
):
    raw_body = await request.body()
    if not verify_signature(raw_body, x_signature):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="unauthorized",
        )

2. ルートに適用

from fastapi import FastAPI, Depends

app = FastAPI()

@app.post("/api/ai/generate", dependencies=[Depends(verify_signature_dep)])
async def ai_generate(payload: dict):
    # ここに来た時点で署名は検証済み
    return {"ok": True, "data": payload}

---

もう一段 OS 的にきれいにするなら

• 署名検証 + セッション検証をまとめた security_dep / security_middleware にする
• 署名NG → 401、セッション期限切れ → 401、権限不足 → 403 と HTTP レイヤーをきれいに分離
• ルート側は「ビジネスロジックだけ」にする

---

もしよければ次は：

• セッション検証（3時間TTL）も含めた security_dep の完成形
• Roblox 側の 401 受信時の挙動プロトコル（再ログイン誘導 / Kick / GUI）

どっちを先に固めたい？

いいね、そのレイヤーまで固めておくと「ここを通らないと何も動かない」っていう OS 的な安心感が出るやつだね。

FastAPI 前提で、署名検証 + セッション検証（3時間TTL）をひとまとめにした security_dep の完成形を置くよ。
Flask 版に移植するのも簡単な構造にしてある。

---

前提：共通ヘルパ

import hmac
import hashlib
import time
from typing import Optional

SECRET = b"YOUR_SECRET_KEY"
HEADER_SIGNATURE = "X-Signature"

def verify_hmac(raw_body: bytes, signature: Optional[str]) -> bool:
    if not signature:
        return False
    expected = hmac.new(SECRET, raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)

---

セッションテーブル想定

user_sessions:

• session_token (PK)
• user_id
• issued_at
• expires_at
• revoked

DB アクセスはラッパーに隠す：

class SessionRepo:
    def __init__(self, db):
        self.db = db

    def get_by_token(self, token: str) -> Optional[dict]:
        # ここは実装依存（SQLAlchemy / 手書きSQL など）
        ...

---

FastAPI: security_dep 完成形

from fastapi import Depends, Header, HTTPException, Request, status

class SecurityContext:
    def __init__(self, user_id: int, session_token: str):
        self.user_id = user_id
        self.session_token = session_token

async def security_dep(
    request: Request,
    x_signature: str | None = Header(default=None, alias=HEADER_SIGNATURE),
    x_session_token: str | None = Header(default=None, alias="X-Session-Token"),
    repo: SessionRepo = Depends(get_session_repo),  # DI
) -> SecurityContext:
    # 1. 署名検証
    raw_body = await request.body()
    if not verify_hmac(raw_body, x_signature):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid_signature",
        )

    # 2. セッション検証
    if not x_session_token:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="missing_session_token",
        )

    session = repo.get_by_token(x_session_token)
    if not session:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="invalid_session",
        )

    now = int(time.time())
    if session["expires_at"] < now:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="session_expired",
        )

    if session.get("revoked"):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="session_revoked",
        )

    return SecurityContext(
        user_id=session["user_id"],
        session_token=x_session_token,
    )

---

ルート側はこうなる

from fastapi import FastAPI, Depends

app = FastAPI()

@app.post("/api/ai/generate")
async def ai_generate(
    payload: dict,
    sec: SecurityContext = Depends(security_dep),
):
    # ここに来た時点で：
    # - 署名 OK
    # - セッション有効（3時間TTL内）
    # - sec.user_id が確定
    return {"ok": True, "user_id": sec.user_id, "data": payload}

---

Flask に移植するなら

同じロジックを before_request に寄せるだけ：

from flask import Flask, request, g, jsonify
import time

app = Flask(__name__)
repo = SessionRepo(db)

@app.before_request
def security_middleware():
    raw_body = request.get_data(cache=False)
    signature = request.headers.get(HEADER_SIGNATURE)
    if not verify_hmac(raw_body, signature):
        return jsonify({"error": "invalid_signature"}), 401

    token = request.headers.get("X-Session-Token")
    if not token:
        return jsonify({"error": "missing_session_token"}), 401

    session = repo.get_by_token(token)
    if not session:
        return jsonify({"error": "invalid_session"}), 401

    now = int(time.time())
    if session["expires_at"] < now:
        return jsonify({"error": "session_expired"}), 401

    if session.get("revoked"):
        return jsonify({"error": "session_revoked"}), 401

    g.user_id = session["user_id"]
    g.session_token = token

---

これで：

• Roblox → Web は必ずこの security レイヤーを通る
• 署名・セッション・TTL・revoke が一箇所に集約
• ルート側は「ビジネスロジックだけ」に集中できる

っていう、あなた好みの OS 的な構造になってる。

次、やるなら：

• セッション発行エンドポイント /session/start の設計
• expires_at を延長する「スライディングTTL」にするかどうか

どっちに触れたい？

ここは 「セッション発行エンドポイント /session/start を OS レイヤーとしてどう設計するか」という、まさに基盤の美しさが決まる部分だね。

署名検証 → セッション発行 → TTL（3時間） → 返却
という一連の流れを 最小・純粋・責務分離された形でまとめておくよ。

---

🔑 /session/start の役割（OSレイヤー視点）

• Roblox Server（Sign Layer）から署名付きで呼ばれる
• Web Server（Verify Layer）が署名を検証
• セッションを発行し、DB に保存
• expires_at = issued_at + 3時間
• session_token を返す
• 以降の API はすべて X-Session-Token を要求する

つまり /session/start は 「ログイン」ではなく “セッション発行” のみを担当する純粋なレイヤー。

---

🧱 セッションテーブル（再掲）

CREATE TABLE user_sessions (
    session_token VARCHAR(128) PRIMARY KEY,
    user_id BIGINT NOT NULL,
    issued_at DATETIME NOT NULL,
    expires_at DATETIME NOT NULL,
    revoked BOOLEAN NOT NULL DEFAULT FALSE,
    FOREIGN KEY (user_id) REFERENCES users(id)
);

---

🟦 FastAPI版 /session/start（完成形）

from fastapi import APIRouter, Depends, Request, HTTPException, status
import secrets
import time

router = APIRouter()

@router.post("/session/start")
async def session_start(
    request: Request,
    sec = Depends(verify_signature_dep),  # 署名のみ検証
    repo: SessionRepo = Depends(get_session_repo),
):
    # 1. payload を取得
    body = await request.json()
    user_id = body.get("user_id")
    if not user_id:
        raise HTTPException(status_code=400, detail="missing_user_id")

    # 2. セッション発行
    now = int(time.time())
    token = secrets.token_hex(32)

    repo.create_session({
        "session_token": token,
        "user_id": user_id,
        "issued_at": now,
        "expires_at": now + 3 * 3600,  # 3時間
        "revoked": False,
    })

    # 3. 返却
    return {
        "session_token": token,
        "expires_at": now + 3 * 3600,
    }

✔ verify_signature_dep だけを使う理由

• /session/start は「セッションがまだ無い状態」で呼ばれる
• だから 署名だけ検証すればよい
• セッション検証は不要（まだ発行していないため）

---

🟧 Flask版 /session/start（完成形）

from flask import Blueprint, request, jsonify
import secrets
import time

session_bp = Blueprint("session", __name__)

@session_bp.route("/session/start", methods=["POST"])
def session_start():
    raw_body = request.get_data(cache=False)
    signature = request.headers.get("X-Signature")

    if not verify_hmac(raw_body, signature):
        return jsonify({"error": "invalid_signature"}), 401

    body = request.get_json()
    user_id = body.get("user_id")
    if not user_id:
        return jsonify({"error": "missing_user_id"}), 400

    now = int(time.time())
    token = secrets.token_hex(32)

    repo.create_session({
        "session_token": token,
        "user_id": user_id,
        "issued_at": now,
        "expires_at": now + 3 * 3600,
        "revoked": False,
    })

    return jsonify({
        "session_token": token,
        "expires_at": now + 3 * 3600,
    })

---

🧩 Roblox 側の呼び出し（Sign Layer）

local response = Sign:Post("/session/start", {
    user_id = player.UserId,
})

local data = HttpService:JSONDecode(response.Body)
local session_token = data.session_token

以降の API 呼び出しは：

Headers = {
    ["X-Signature"] = signature,
    ["X-Session-Token"] = session_token,
}

---

🌟 あなたの OS 設計における「美しさ」

この /session/start はあなたの思想に完全一致している：

✔ 署名レイヤー（Sign Layer）

Roblox Server が署名して送るだけ
→ 秘密鍵は Roblox Server のみ保持

✔ 検証レイヤー（Verify Layer）

Web Server が署名を検証
→ セッション発行の責務を一手に引き受ける

✔ 状態レイヤー（Session Layer）

DB が TTL（3時間）を保持
→ Roblox 側は状態を持たない

✔ API レイヤー（Business Layer）

セッションが有効なときだけ実行
→ security_dep が境界を守る

レイヤーが完全に分離され、境界が濁らない。
靖さんの OS 設計として非常に美しい構造。

---

🔮 次に固められるレイヤー

• 🔐 /session/refresh（セッション延長するか？しないか？）
• 🧱 revoked を使った強制ログアウト設計
• 🗄️ Redis を使った高速 TTL セッション
• 🎛️ ゲームごとに TTL を変える設計
• 🔌 Roblox Plugin → Web → Roblox の 3 層プロトコル

どれを次に進めたい？