iT邦幫忙

2026 iThome 鐵人賽
DAY 27
0
AI Engineering

30天用 Claude Code + LangGraph 實作個人化 AI 學習教練系列 第 27 篇

Day 27：登入與安全（簡化版） - 保護用戶資料

18th鐵人賽
50 瀏覽

  • 分享至 

  • xImage
    •  

到目前為止，系統的 API 都沒有認證。任何人只要帶上任意 user_id，就能查到別人的計畫和進度，這是很嚴重的漏洞。用戶 1 的學習計畫，用戶 2 不應該看得到。今天要加上認證層，讓系統知道請求來自誰，以及誰能存取哪些資料。

做法是用 JWT（JSON Web Token）做無狀態認證。用戶登入後會拿到一個 token，之後每次請求都要帶上它，伺服器則驗證 token 是否有效。密碼用 bcrypt 雜湊後儲存，即使資料庫外洩，也看不到明文密碼。為了防止暴力破解，每個 IP 在短時間內最多嘗試 5 次登入。

概念：JWT 是自己簽名的紙條

想像你去電影院買票。售票員給你一張紙條：「某某人，2026 年 10 月 9 日，晚上 8:00，第三廳」。這張紙條上寫著重要資訊，還有售票員的簽名。之後進場時，檢票員不必查資料庫，只要確認簽名沒有被改過，就能判斷你能不能進場。

JWT 就是這樣一張紙條。它包含三個部分：

header.payload.signature
  • header：說明這是 JWT，用什麼演算法簽名
  • payload：實際的資料內容（user_id、過期時間等）
  • signature：伺服器的簽名，用密鑰生成

任何人都能讀取 payload，因為它只是 base64 編碼，並沒有加密。但只有掌握密鑰的伺服器能產生正確的簽名。如果有人改了 payload，簽名就會對不上。

好處是：

  • 無狀態：伺服器不用存 token 清單，驗證 token 時直接檢查簽名
  • 可攜帶：token 裡面帶著 user_id，不用每次都查資料庫
  • 安全：只要密鑰不洩露，token 就不能被篡改
登入流程：
用戶輸入帳號密碼
        ↓
伺服器檢查密碼（bcrypt 比對）
        ↓ 通過
生成 JWT token
        ↓
回傳 token 給用戶
        ↓
用戶每次請求都帶著 token
        ↓
伺服器驗證 token 的簽名和過期時間
        ↓ 有效
執行操作

實作步驟

步驟 1：安裝必要套件

pip install python-jose cryptography passlib
  • python-jose：JWT 的實現
  • cryptography：用於 JWT 簽名
  • passlib：密碼雜湊工具

步驟 2：更新 User 模型

檔案位置: backend/models.py
狀態: 修改檔案（在 User 類別裡）
用途: 加上密碼欄位和最後登入時間

class User(Base):
    """用戶模型"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, nullable=False, index=True)
    password_hash = Column(String, nullable=False)  # 不存明文密碼，只存雜湊
    created_at = Column(DateTime, default=datetime.datetime.utcnow, nullable=False)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
    
    # 反向參照
    profiles = relationship("Profile", back_populates="user", cascade="all, delete-orphan")
    goals = relationship("Goal", back_populates="user", cascade="all, delete-orphan")
    plans = relationship("Plan", back_populates="user", cascade="all, delete-orphan")
    conversations = relationship("Conversation", back_populates="user", cascade="all, delete-orphan")
    notifications = relationship("Notification", back_populates="user", cascade="all, delete-orphan")

步驟 3：定義認證相關的 Schema

檔案位置: backend/schemas.py
狀態: 修改檔案（接在檔案最後面）
用途: 定義登入、註冊、token 的格式

class RegisterRequest(BaseModel):
    """註冊請求"""
    email: str
    password: str


class LoginRequest(BaseModel):
    """登入請求"""
    email: str
    password: str


class TokenResponse(BaseModel):
    """Token 回應"""
    access_token: str
    token_type: str = "bearer"
    expires_in: int  # 秒數


class TokenPayload(BaseModel):
    """Token 內容"""
    sub: int  # user_id
    exp: int  # 過期時間戳
    iat: int  # 發行時間戳

步驟 4：建立認證工具模組

檔案位置: backend/auth.py
狀態: 新建檔案
用途: 集中管理 JWT、密碼雜湊等邏輯

from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from models import User
from schemas import TokenPayload
from database import get_session

# JWT 設定
SECRET_KEY = "your-secret-key-change-this-in-production"  # ⚠️ 正式環境要改成安全的
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60  # token 有效期 60 分鐘

# 密碼雜湊設定
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# HTTP Bearer token 提取器
security = HTTPBearer()


def hash_password(password: str) -> str:
    """將密碼雜湊"""
    return pwd_context.hash(password)


def verify_password(plain_password: str, hashed_password: str) -> bool:
    """驗證密碼是否正確"""
    return pwd_context.verify(plain_password, hashed_password)


def create_access_token(user_id: int, expires_delta: Optional[timedelta] = None) -> str:
    """建立 JWT token"""
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    to_encode = {
        "sub": user_id,
        "exp": expire,
        "iat": datetime.utcnow()
    }
    
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def verify_token(token: str) -> TokenPayload:
    """驗證 JWT token，返回 token 內容"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Token 無效"
            )
        return TokenPayload(
            sub=user_id,
            exp=payload.get("exp"),
            iat=payload.get("iat")
        )
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token 驗證失敗"
        )


def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    session: Session = Depends(get_session)
) -> User:
    """作為 Depends，用來保護 API 端點
    
    使用方式：
    @app.get("/protected")
    async def protected_endpoint(current_user: User = Depends(get_current_user)):
        return {"user_id": current_user.id}
    """
    token = credentials.credentials
    payload = verify_token(token)
    
    user = session.query(User).filter(User.id == payload.sub).first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="用戶不存在"
        )
    
    return user

步驟 5：實現登入和註冊 API

檔案位置: backend/main.py
狀態: 修改檔案（在現有端點最上面加 import）
用途: 提供認證端點

先加 import：

from auth import hash_password, verify_password, create_access_token, get_current_user, ACCESS_TOKEN_EXPIRE_MINUTES
from datetime import timedelta
from schemas import RegisterRequest, LoginRequest, TokenResponse
from fastapi import HTTPException, status

再加上端點（放在檔案最後）：

# 記錄登入嘗試，用於 rate limiting
login_attempts = {}  # {"ip": [時間戳, 時間戳, ...]}


def get_client_ip(request: Request) -> str:
    """取得客戶端 IP"""
    return request.client.host if request.client else "unknown"


def check_rate_limit(ip: str, max_attempts: int = 5, window_seconds: int = 300) -> bool:
    """檢查 IP 是否超過登入次數限制
    
    Args:
        ip: 客戶端 IP
        max_attempts: 時間窗口內允許的最大嘗試次數
        window_seconds: 時間窗口（秒）
    
    Returns:
        True 表示允許，False 表示超過限制
    """
    now = datetime.utcnow().timestamp()
    
    if ip not in login_attempts:
        login_attempts[ip] = []
    
    # 清除超過時間窗口的嘗試記錄
    login_attempts[ip] = [ts for ts in login_attempts[ip] if now - ts < window_seconds]
    
    if len(login_attempts[ip]) >= max_attempts:
        return False
    
    login_attempts[ip].append(now)
    return True


@app.post("/auth/register", response_model=TokenResponse)
async def register(
    request: Request,
    data: RegisterRequest,
    session: Session = Depends(get_session)
):
    """註冊新用戶"""
    # 檢查 IP 限制
    ip = get_client_ip(request)
    if not check_rate_limit(ip):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="登入嘗試過於頻繁，請稍後再試"
        )
    
    # 檢查 email 是否已存在
    existing_user = session.query(User).filter(User.email == data.email).first()
    if existing_user:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="此信箱已被使用"
        )
    
    # 驗證密碼強度（簡單版：至少 6 字元）
    if len(data.password) < 6:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="密碼至少 6 個字元"
        )
    
    # 建立新用戶
    hashed_password = hash_password(data.password)
    new_user = User(email=data.email, password_hash=hashed_password)
    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    
    # 立即發行 token
    access_token = create_access_token(new_user.id)
    return TokenResponse(
        access_token=access_token,
        token_type="bearer",
        expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )


@app.post("/auth/login", response_model=TokenResponse)
async def login(
    request: Request,
    data: LoginRequest,
    session: Session = Depends(get_session)
):
    """登入用戶"""
    # 檢查 IP 限制
    ip = get_client_ip(request)
    if not check_rate_limit(ip):
        raise HTTPException(
            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
            detail="登入嘗試過於頻繁，請稍後再試"
        )
    
    # 查詢用戶
    user = session.query(User).filter(User.email == data.email).first()
    if not user or not verify_password(data.password, user.password_hash):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="信箱或密碼錯誤"
        )
    
    # 發行 token
    access_token = create_access_token(user.id)
    return TokenResponse(
        access_token=access_token,
        token_type="bearer",
        expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )


@app.post("/auth/refresh", response_model=TokenResponse)
async def refresh_token(
    current_user: User = Depends(get_current_user),
):
    """刷新 token"""
    access_token = create_access_token(current_user.id)
    return TokenResponse(
        access_token=access_token,
        token_type="bearer",
        expires_in=ACCESS_TOKEN_EXPIRE_MINUTES * 60
    )


@app.get("/auth/me")
async def get_me(current_user: User = Depends(get_current_user)):
    """查詢目前登入的用戶信息"""
    return {
        "id": current_user.id,
        "email": current_user.email,
        "created_at": current_user.created_at.isoformat()
    }

檔案最上面還要加上 Request 的 import：

from fastapi import Request

步驟 6：保護現有 API

檔案位置: backend/main.py
狀態: 修改檔案（更新現有的 API 端點）
用途: 讓現有的 API 都檢查 token

以 GET /dashboard 為例，改成：

@app.get("/dashboard")
async def get_dashboard(
    current_user: User = Depends(get_current_user),
    session: Session = Depends(get_session)
):
    """取得儀表板資訊（已認證用戶）"""
    # 後面的邏輯不變，只是加上了 current_user 這個參數
    # 所有查詢都改成用 current_user.id 代替 user_id 參數
    user_id = current_user.id
    # 原本的邏輯...

凡是涉及用戶資料的 API，都要加上 current_user: User = Depends(get_current_user) 參數，並用 current_user.id 取代直接從請求中讀取的 user_id。

步驟 7：更新前端登入頁

檔案位置: frontend/app/login/page.tsx
狀態: 新建/修改檔案
用途: 讓用戶登入

'use client';

import { useState } from 'react';
import { useRouter } from 'next/navigation';

export default function LoginPage() {
  const [email, setEmail] = useState('');
  const [password, setPassword] = useState('');
  const [error, setError] = useState('');
  const [loading, setLoading] = useState(false);
  const router = useRouter();

  const handleLogin = async (e: React.FormEvent) => {
    e.preventDefault();
    setError('');
    setLoading(true);

    try {
      const response = await fetch(`${process.env.NEXT_PUBLIC_API_URL}/auth/login`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ email, password }),
      });

      if (!response.ok) {
        const data = await response.json();
        setError(data.detail || '登入失敗');
        return;
      }

      const data = await response.json();
      // 存 token 到 localStorage
      localStorage.setItem('access_token', data.access_token);
      // 導向儀表板
      router.push('/dashboard');
    } catch (err) {
      setError('登入出錯，請稍後再試');
    } finally {
      setLoading(false);
    }
  };

  const handleRegister = async (e: React.FormEvent) => {
    e.preventDefault();
    setError('');
    setLoading(true);

    try {
      const response = await fetch(`${process.env.NEXT_PUBLIC_API_URL}/auth/register`, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ email, password }),
      });

      if (!response.ok) {
        const data = await response.json();
        setError(data.detail || '註冊失敗');
        return;
      }

      const data = await response.json();
      localStorage.setItem('access_token', data.access_token);
      router.push('/dashboard');
    } catch (err) {
      setError('註冊出錯，請稍後再試');
    } finally {
      setLoading(false);
    }
  };

  return (
    <div className="min-h-screen flex items-center justify-center bg-gray-50">
      <div className="max-w-md w-full space-y-8">
        <div>
          <h2 className="text-center text-3xl font-bold">AI 學習教練</h2>
          <p className="mt-2 text-center text-gray-600">登入或建立新帳號</p>
        </div>

        {error && <div className="bg-red-50 text-red-700 p-3 rounded">{error}</div>}

        <form onSubmit={handleLogin} className="space-y-4">
          <input
            type="email"
            placeholder="信箱"
            value={email}
            onChange={(e) => setEmail(e.target.value)}
            className="w-full px-4 py-2 border rounded"
            required
          />
          <input
            type="password"
            placeholder="密碼"
            value={password}
            onChange={(e) => setPassword(e.target.value)}
            className="w-full px-4 py-2 border rounded"
            required
          />
          <button
            type="submit"
            disabled={loading}
            className="w-full bg-blue-600 text-white py-2 rounded hover:bg-blue-700"
          >
            {loading ? '登入中...' : '登入'}
          </button>
        </form>

        <div className="relative">
          <div className="absolute inset-0 flex items-center">
            <div className="w-full border-t"></div>
          </div>
          <div className="relative flex justify-center text-sm">
            <span className="px-2 bg-gray-50 text-gray-500">或</span>
          </div>
        </div>

        <button
          onClick={(e) => handleRegister(e)}
          disabled={loading}
          className="w-full bg-green-600 text-white py-2 rounded hover:bg-green-700"
        >
          {loading ? '註冊中...' : '建立新帳號'}
        </button>
      </div>
    </div>
  );
}

步驟 8：前端 API 呼叫時帶上 token

檔案位置: frontend/lib/api.ts
狀態: 新建/修改檔案
用途: 為所有 API 呼叫自動帶上 token

const API_URL = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000';

export async function apiCall(
  endpoint: string,
  options: RequestInit = {}
) {
  const token = localStorage.getItem('access_token');
  
  const headers: HeadersInit = {
    'Content-Type': 'application/json',
    ...options.headers,
  };

  if (token) {
    headers['Authorization'] = `Bearer ${token}`;
  }

  const response = await fetch(`${API_URL}${endpoint}`, {
    ...options,
    headers,
  });

  // 如果 token 過期（401），清除並導向登入
  if (response.status === 401) {
    localStorage.removeItem('access_token');
    window.location.href = '/login';
  }

  return response;
}

// 使用範例
export async function getDashboard() {
  const response = await apiCall('/dashboard');
  return response.json();
}

測試認證系統

測試 1：註冊和登入

# 1. 註冊新用戶
curl -X POST http://localhost:8000/auth/register \
  -H "Content-Type: application/json" \
  -d '{"email":"test@example.com","password":"password123"}'

# 回應應該包含 access_token
# {"access_token":"eyJ...","token_type":"bearer","expires_in":3600}

# 2. 用錯密碼登入
curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email":"test@example.com","password":"wrong"}'

# 應該回傳 401 錯誤

# 3. 用正確密碼登入
curl -X POST http://localhost:8000/auth/login \
  -H "Content-Type: application/json" \
  -d '{"email":"test@example.com","password":"password123"}'

# 獲得 token

測試 2：使用 token 訪問保護的端點

# 取得 token（從上面的登入回應）
TOKEN="eyJ..."

# 訪問受保護的端點
curl http://localhost:8000/auth/me \
  -H "Authorization: Bearer $TOKEN"

# 應該回傳用戶資訊
# {"id":1,"email":"test@example.com","created_at":"2026-10-09T..."}

# 試著不帶 token 訪問
curl http://localhost:8000/auth/me

# 應該回傳 403 錯誤

測試 3：Rate limiting

# 連續登入 6 次（超過限制的 5 次）
for i in {1..6}; do
  curl -X POST http://localhost:8000/auth/login \
    -H "Content-Type: application/json" \
    -d '{"email":"test@example.com","password":"wrong"}'
done

# 第 6 次應該回傳 429 Too Many Requests

常見問題

SECRET_KEY 寫在程式碼裡

現在 SECRET_KEY 是直接寫死在程式碼裡的字串。如果這份程式碼上傳到 GitHub，任何人都能看到金鑰，進而偽造 token。

解決：用環境變數：

import os
SECRET_KEY = os.getenv("SECRET_KEY", "dev-key-only-for-testing")

正式環境設定一個長、亂、隨機的密鑰：

# 在伺服器上
export SECRET_KEY=$(openssl rand -hex 32)

忘記更新所有 API，某些端點還沒加認證

用戶登入後，仍能在 URL 裡直接改 user_id，看到別人的資料。

解決：用 grep 檢查所有涉及 user_id 的 endpoint，逐一加上 Depends(get_current_user) 並改成用 current_user.id。

###前端登入後沒存 token

用戶登入成功，卻看不到儀表板，因為後續請求都沒有帶上 token。

解決：登入成功後，一定要 localStorage.setItem('access_token', token)。

Token 過期但用戶不知道

用戶正在使用系統時 token 過期，API 回傳 401，但頁面沒有任何反應。

解決：在 apiCall 函式裡檢查 401，自動清除 token 並導向登入頁。

下一步 Day 28 要做單元測試，確認整個系統的邏輯正確。認證涉及安全，程式碼特別需要測試，一個小 bug 就可能讓整個系統外洩。


上一篇
Day 26：通知與提醒（簡化版） - 自動化提醒
下一篇
Day 28：基本單元測試 - 自動化測試確保應用品質
系列文
30天用 Claude Code + LangGraph 實作個人化 AI 學習教練 共 30 篇
  1. 26
    Day 26：通知與提醒（簡化版） - 自動化提醒
  2. 27
    Day 27：登入與安全（簡化版） - 保護用戶資料
  3. 28
    Day 28：基本單元測試 - 自動化測試確保應用品質
  4. 29
    Day 29：部署上線（簡化版） - 從本地到生產環境
  5. 30
    Day 30：發表與收尾 - 從 Prototype 到 Product
完整目錄
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言
iThome鐵人賽
參賽組數
929 組
團體組數
32 組
累計文章數
20132 篇
完賽人數
534 人
iT+看影片追技術 看影片追技術 看更多
熱門tag
15th鐵人賽 16th鐵人賽 13th鐵人賽 18th鐵人賽 17th鐵人賽 14th鐵人賽 12th鐵人賽 11th鐵人賽 鐵人賽 2019鐵人賽 javascript 2018鐵人賽 python 2017鐵人賽 windows php c# linux windows server css
熱門問題
熱門回答
熱門文章
IT邦幫忙