FastAPI: JWTを使ったログインシステム

Last updated at 2025-04-13Posted at 2025-04-13

バックエンド

main.py

#
#	main.py
#
#					Apr/13/2025
#
# ------------------------------------------------------------------
import sys
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional
from fastapi.middleware.cors import CORSMiddleware

# ------------------------------------------------------------------
# シークレットキーとアルゴリズム
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# パスワードのハッシュ化設定
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# OAuth2 用のスキーム（トークン取得用のURL）
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],  # フロントエンドのURLに制限するのが望ましい
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
)

# 仮のユーザーデータベース
fake_users_db = {
	"scott": {
		"username": "scott",
		"full_name": "Scott Wonderland",
		"hashed_password": pwd_context.hash("scott123"),
		"email": "scott@example.net",
		"disabled": False,
	},
	"jack": {
		"username": "jack",
		"full_name": "Jack Smith",
		"hashed_password": pwd_context.hash("jack123"),
		"email": "jack@example.net",
		"disabled": False,
	},
	"betty": {
		"username": "betty",
		"full_name": "Betty White",
		"hashed_password": pwd_context.hash("betty123"),
		"email": "betty@example.net",
		"disabled": False,
	}
}

# ------------------------------------------------------------------
# ユーザーモデル
class User:
	def __init__(self, username: str, full_name: Optional[str] = None, \
		email: Optional[str] = None, disabled: Optional[bool] = None):
		self.username = username
		self.full_name = full_name
		self.email = email
		self.disabled = disabled

class UserInDB(User):
	def __init__(self, username: str, hashed_password: str, **kwargs):
		super().__init__(username, **kwargs)
		self.hashed_password = hashed_password

# ------------------------------------------------------------------
# ユーザー取得
def get_user(db, username: str):
	if username in db:
		user_dict = db[username]
		return UserInDB(**user_dict)

# パスワード検証
def verify_password(plain_password, hashed_password):
	return pwd_context.verify(plain_password, hashed_password)

# ユーザー認証
def authenticate_user(db, username: str, password: str):
	user = get_user(db, username)
	if not user or not verify_password(password, user.hashed_password):
		return False
	return user

# ------------------------------------------------------------------
# アクセストークンの作成
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
	to_encode = data.copy()
	expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
	to_encode.update({"exp": expire})
	return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# トークンからユーザーを取得
async def get_current_user(token: str = Depends(oauth2_scheme)):
	credentials_exception = HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED,
		detail="認証に失敗しました",
		headers={"WWW-Authenticate": "Bearer"},
	)
	try:
		payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
		username: str = payload.get("sub")
		if username is None:
			raise credentials_exception
	except JWTError:
		raise credentials_exception
	user = get_user(fake_users_db, username)
	if user is None:
		raise credentials_exception
	return user

# ユーザーが有効かどうか確認
async def get_current_active_user(current_user: User = Depends(get_current_user)):
	if current_user.disabled:
		raise HTTPException(status_code=400, detail="無効なユーザー")
	return current_user

# ------------------------------------------------------------------
# トークン取得エンドポイント
@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
	sys.stderr.write("*** token *** aaa ***\n")
	user = authenticate_user(fake_users_db, form_data.username, form_data.password)
	if not user:
		raise HTTPException(status_code=400, detail="ユーザー名またはパスワードが無効です")
	access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
	access_token = create_access_token(
		data={"sub": user.username}, expires_delta=access_token_expires
	)
	sys.stderr.write("*** token *** end ***\n")
	return {"access_token": access_token, "token_type": "bearer"}

# ------------------------------------------------------------------
# 保護されたエンドポイント
@app.get("/users/me")
async def read_users_me(current_user: User = Depends(get_current_active_user)):
	sys.stderr.write("*** read_users_me ***\n")
	return {"username": current_user.username, \
	"full_name": current_user.full_name, \
	"email": current_user.email}

# ------------------------------------------------------------------

サーバーの起動

uvicorn main:app --reload

フロントエンド

index.html

<!DOCTYPE html>
<html lang="ja">
<head>
    <meta charset="UTF-8">
    <title>JWT ログイン</title>
</head>
<body>
    <h1>ログイン</h1>
    <form id="login-form">
        <label for="username">ユーザー名:</label><br>
        <input type="text" id="username" name="username"><br>
        <label for="password">パスワード:</label><br>
        <input type="password" id="password" name="password"><br><br>
        <button type="submit">ログイン</button>
    </form>

    <h2>ユーザー情報</h2>
    <button onclick="getUserInfo()">ユーザー情報取得</button>
    <pre id="user-info"></pre>

    <script src="index.js"></script>
</body>
</html>

index.js

        let accessToken = null

        document.getElementById("login-form").addEventListener("submit", async (e) => {
            e.preventDefault()
            const username = document.getElementById("username").value
            const password = document.getElementById("password").value

            const response = await fetch("http://localhost:8000/token", {
                method: "POST",
                headers: {
                    "Content-Type": "application/x-www-form-urlencoded",
                },
                body: new URLSearchParams({
                    "username": username,
                    "password": password
                }),
            })

            const data = await response.json()
            if (response.ok) {
                accessToken = data.access_token
                alert("ログイン成功！")
            } else {
                alert("ログイン失敗: " + (data.detail || "不明なエラー"))
            }
        })

        async function getUserInfo() {
            if (!accessToken) {
                alert("先にログインしてください")
                return
            }

            const response = await fetch("http://localhost:8000/users/me", {
                headers: {
                    "Authorization": "Bearer " + accessToken
                }
            })

            const data = await response.json()
            document.getElementById("user-info").textContent = JSON.stringify(data, null, 2)
        }

Web サーバーの起動

http-server

ブラウザーでアクセス

ユーザー名　scott
パスワード　scott123
でログイン

ユーザー情報取得

検証スクリプト

ログイン成功

http --form POST http://localhost:8000/token username="scott" password="scott123"

$ http --form POST http://localhost:8000/token username="scott" password="scott123"
HTTP/1.1 200 OK
content-length: 165
content-type: application/json
date: Sun, 13 Apr 2025 05:23:17 GMT
server: uvicorn

{
    "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJzY290dCIsImV4cCI6MTc0NDUyMzU5OH0.TXxpZNKVCveAvQKqpgCcfJObapzE4LCW8KvKAOpssn0",
    "token_type": "bearer"
}

ログイン失敗

http --form POST http://localhost:8000/token username="scott" password="scott124"

$ http --form POST http://localhost:8000/token username="scott" password="scott124"
HTTP/1.1 400 Bad Request
content-length: 67
content-type: application/json
date: Sun, 13 Apr 2025 05:24:04 GMT
server: uvicorn

{
    "detail": "ユーザー名またはパスワードが無効です"
}

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up