在現代 Web 開發中，API 安全性至關重要。本文將引導您逐步強化 FastAPI 應用程式的安全性，涵蓋使用者驗證和授權的實作。我們將使用 OAuth2 和 JWT 技術來保護 API 端點，確保只有經過驗證的使用者才能存取受保護的資源。此外，我們還將探討如何使用 pytest 框架撰寫單元測試，以驗證應用程式功能的正確性，並示範如何組態 CORS 中介軟體來處理跨來源請求。透過這些技術，您可以有效提升 FastAPI 應用程式的安全性，並確保其穩定可靠地執行。
強化 FastAPI 應用程式的安全性 - 使用者驗證與授權實作
在前面的程式碼區塊中，函式接收 token 字串作為引數，並在 try 區塊中執行多項檢查。首先檢查 token 的過期時間，如果沒有過期時間，則表示未提供 token。其次檢查 token 的有效性，若 token 已過期，則丟擲例外通知使用者。若 token 有效，則傳回解碼後的 payload。
在 except 區塊中，針對任何 JWT 錯誤丟擲錯誤請求例外。
內容解密：
- 函式目的：驗證傳送給應用程式的 token。
- 檢查順序：先檢查過期時間，再驗證 token 有效性。
- 錯誤處理：對於 JWT 錯誤，統一在 except 區塊處理。
處理使用者驗證
我們已經成功實作了密碼雜湊與比較、JWT 建立與解碼等元件。現在來建立將被注入事件路由的依賴函式，作為檢索活躍會話使用者的唯一依據。
在 auth/authenticate.py 中新增以下程式碼：
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from auth.jwt_handler import verify_access_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/user/signin")
async def authenticate(token: str = Depends(oauth2_scheme)) -> str:
if not token:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Sign in for access"
)
decoded_token = verify_access_token(token)
return decoded_token["user"]
內容解密：
oauth2_scheme定義：指定 OAuth2 密碼流的 token URL。authenticate函式邏輯：檢查 token 存在性，驗證 token 有效性，並傳回解碼後的使用者資訊。- 依賴注入：
oauth2_scheme被注入authenticate函式。
更新應用程式
更新使用者登入路由
在 routes/users.py 中更新匯入：
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from auth.jwt_handler import create_access_token
from models.users import User, TokenResponse
更新 sign_user_in() 路由函式：
async def sign_user_in(user: OAuth2PasswordRequestForm = Depends()) -> dict:
user_exist = await User.find_one(User.email == user.username)
# ...
if hash_password.verify_hash(user.password, user_exist.password):
access_token = create_access_token(user_exist.email)
return {
"access_token": access_token,
"token_type": "Bearer"
}
內容解密：
OAuth2PasswordRequestForm使用：確保嚴格遵循 OAuth 規範。- 密碼驗證與 token 生成：驗證密碼正確後生成 access token。
- 回應模型更新：使用
TokenResponse模型傳回 token 資訊。
更新事件路由以支援授權
在事件路由中注入 authenticate 依賴：
from auth.authenticate import authenticate
async def create_event(body: Event, user: str = Depends(authenticate)) -> dict:
# ...
async def update_event(id: PydanticObjectId, body: EventUpdate, user: str = Depends(authenticate)) -> Event:
# ...
async def delete_event(id: PydanticObjectId, user: str = Depends(authenticate)) -> dict:
# ...
內容解密：
- 依賴注入：將
authenticate函式注入事件路由，確保只有授權使用者可以操作。 - 自動更新檔案：FastAPI 自動更新互動式檔案以顯示受保護的路由。
此圖示說明瞭如何使用 OAuth2PasswordBearer 和依賴注入來保護 FastAPI 路由：
@startuml
skinparam backgroundColor #FEFEFE
skinparam sequenceArrowThickness 2
title FastAPI 安全性強化與測試實作
actor "客戶端" as client
participant "API Gateway" as gateway
participant "認證服務" as auth
participant "業務服務" as service
database "資料庫" as db
queue "訊息佇列" as mq
client -> gateway : HTTP 請求
gateway -> auth : 驗證 Token
auth --> gateway : 認證結果
alt 認證成功
gateway -> service : 轉發請求
service -> db : 查詢/更新資料
db --> service : 回傳結果
service -> mq : 發送事件
service --> gateway : 回應資料
gateway --> client : HTTP 200 OK
else 認證失敗
gateway --> client : HTTP 401 Unauthorized
end
@enduml此圖示顯示了保護路由的流程，使用者必須透過驗證才能存取相關資源。
保護FastAPI應用程式的安全
在成功登入後，我們可以建立一個新的活動：
使用命令列建立新活動
首先，讓我們從命令列取得存取令牌：
$ curl -X 'POST' \
'http://0.0.0.0:8080/user/signin' \
-H 'accept: application/json' \
-H 'Content-Type: application/x-www-form-urlencoded' \
-d 'grant_type=&username=reader%40packt.com&password=exemplary&scope=&client_id=&client_secret='
內容解密：
此命令向 /user/signin 端點傳送 POST 請求以取得存取令牌。其中，-H 引數用於設定請求頭，-d 引數用於設定請求體。請求成功後，伺服器將傳回一個 JWT 令牌和令牌型別。
伺服器傳回的回應如下：
{
"access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE2NTA4MjkxODMuNTg3NjAyfQ.MOXjI5GXnyzGNftdlxDGyM119_L11uPq8yCxBHepf04",
"token_type": "Bearer"
}
接下來，讓我們使用存取令牌在命令列中建立一個新活動：
$ curl -X 'POST' \
'http://0.0.0.0:8080/event/new' \
-H 'accept: application/json' \
-H 'Authorization: Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VyIjoicmVhZGVyQHBhY2t0LmNvbSIsImV4cGlyZXMiOjE2NTA4MjkxODMuNTg3NjAyfQ.MOXjI5GXnyzGNftdlxDGyM119_L11uPq8yCxBHepf04' \
-H 'Content-Type: application/json' \
-d '{
"title": "FastAPI Book Launch CLI",
"image": "https://linktomyimage.com/image.png",
"description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
"tags": [
"python",
"fastapi",
"book",
"launch"
],
"location": "Google Meet"
}'
內容解密：
此命令向 /event/new 端點傳送 POST 請求以建立新活動。Authorization: Bearer 請求頭攜帶了存取令牌，用於驗證使用者身份。請求成功後，伺服器傳回以下回應：
{
"message": "Event created successfully"
}
如果未攜帶有效的存取令牌，則會傳回 HTTP 401 Unauthorized 錯誤。
更新受保護的路由
現在，我們已經成功保護了路由，接下來需要更新受保護的路由以滿足以下需求：
- POST 路由：將建立的活動新增到使用者擁有的活動列表中。
- UPDATE 路由：修改路由以確保只有使用者建立的活動才能被更新。
- DELETE 路由：修改路由以確保只有使用者建立的活動才能被刪除。
更新活動檔案類別和路由
首先，在 models/events.py 中為 Event 檔案類別新增 creator 欄位：
class Event(Document):
creator: Optional[str]
內容解密：
此欄位用於記錄建立活動的使用者，以便限制對活動的操作。
接下來，修改 routes/events.py 中的 POST 路由以更新 creator 欄位：
@event_router.post("/new")
async def create_event(body: Event, user: str = Depends(authenticate)) -> dict:
body.creator = user
await event_database.save(body)
return {
"message": "Event created successfully"
}
內容解密：
此路由函式將當前使用者的電子郵件地址作為活動的建立者，並將其儲存到資料函式庫中。
然後，更新 UPDATE 路由：
@event_router.put("/{id}", response_model=Event)
async def update_event(id: PydanticObjectId, body: EventUpdate, user: str = Depends(authenticate)) -> Event:
event = await event_database.get(id)
if event.creator != user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Operation not allowed"
)
內容解密：
此路由函式檢查當前使用者是否為活動的建立者，如果不是，則引發 HTTP 400 Bad Request 例外。
最後，更新 DELETE 路由：
@event_router.delete("/{id}")
async def delete_event(id: PydanticObjectId, user: str = Depends(authenticate)):
event = await event_database.get(id)
if event.creator != user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Operation not allowed"
)
內容解密：
此路由函式同樣檢查當前使用者是否為活動的建立者，如果不是，則引發 HTTP 400 Bad Request 例外。
測試 FastAPI 應用程式
在前一章中，我們學習瞭如何使用 OAuth 和 JSON Web Token（JWT）來保護 FastAPI 應用程式。我們成功地實作了一個認證系統，並且瞭解了依賴注入（Dependency Injection）的概念。我們也學習瞭如何將依賴注入路由中，以限制未授權的存取和操作。到目前為止，我們已經成功地建立了一個具有資料函式庫支援且能夠輕鬆執行 CRUD 操作的安全網頁 API。在本章中，我們將學習什麼是測試，以及如何撰寫測試以確保我們的應用程式行為符合預期。
為什麼需要測試？
測試是應用程式開發週期中不可或缺的一部分。進行應用程式測試的目的是確保應用程式的正確運作狀態，並在佈署到生產環境之前輕鬆檢測出異常情況。雖然在前幾章中，我們一直在手動測試應用程式的端點，但在本章中，我們將學習如何自動化這些測試。
單元測試與 pytest
單元測試是一種測試方法，旨在驗證應用程式中的個別單元（例如函式或方法）是否按照預期工作。pytest 是一個流行的 Python 測試框架，能夠幫助我們輕鬆地撰寫和執行測試。
設定測試環境
首先，我們需要設定測試環境。這包括安裝必要的測試函式庫、組態測試資料函式庫，以及建立測試客戶端。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
測試範例：事件刪除
接下來，讓我們以事件刪除為例，示範如何撰寫測試。首先，我們需要建立一個測試事件，然後使用 TestClient 來模擬刪除請求。
def test_delete_event():
# 建立測試事件
event_id = "6265a83fc823a3c912830074"
# 使用 TestClient 模擬刪除請求
response = client.delete(f"/event/{event_id}", headers={"Authorization": "Bearer YOUR_TOKEN"})
# 驗證回應狀態碼
assert response.status_code == 200
# 驗證回應內容
assert response.json()["message"] == "Event deleted successfully."
內容解密：
- 建立測試事件：在這個步驟中，我們需要先在資料函式庫中建立一個測試事件，以便進行刪除測試。
- 使用 TestClient 模擬刪除請求：我們使用
TestClient來模擬一個 DELETE 請求到/event/{event_id}端點，並傳遞必要的授權標頭。 - 驗證回應狀態碼：我們檢查回應的狀態碼是否為 200，表示請求成功。
- 驗證回應內容：最後，我們驗證回應內容是否包含預期的訊息，即 “Event deleted successfully."。
CORS 組態
在進行測試之前，我們需要確保我們的 FastAPI 應用程式組態了 CORS 中介軟體，以允許跨來源請求。
from fastapi.middleware.cors import CORSMiddleware
origins = ["*"]
app.add_middleware(
CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
內容解密：
- 匯入 CORSMiddleware：首先，我們從
fastapi.middleware.cors匯入CORSMiddleware類別。 - 定義允許的來源：我們定義了一個
origins列表，其中包含允許存取我們 API 的來源。在這個例子中，我們允許所有來源（*）。 - 新增 CORS 中介軟體：使用
add_middleware方法將CORSMiddleware新增到我們的 FastAPI 應用程式中，並組態相關引數。
