FastAPI 结合 JWT

FastAPI 结合 JWT

   JWT（JSON Web Token）是一种基于 JSON 的开放标准（RFC 7519），用于在各方之间传递安全可靠的信息。JWT 可以签名（使用 HMAC 算法或 RSA 等），从而可以验证内容是否被窜改。JWT 通常用于认证和授权流程。
  步调

   在 FastAPI 中，JWT 主要用于保护 API 路由，使其只答应颠末身份验证的用户访问。认证流程大致如下：
  

• **用户登录：**用户通过提交用户名和密码获取 JWT。
  
• **获取 Token：**服务器验证用户凭据后，生成并返回 JWT 给用户。
  
• **访问受保护的路由：**用户在访问受保护的路由时，必要在请求头中携带该 JWT。
  
• **Token 验证：**服务器验证 JWT 的正当性和有效性，答应或拒绝访问受保护资源。
  

安装

1. pip install fastapi uvicorn pyjwt

复制代码

步调

导入必要的模块

1. from fastapi import FastAPI, Depends, HTTPException, status
2. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
3. from pydantic import BaseModel
4. from datetime import datetime, timedelta, timezone
5. import jwt

复制代码

设置配置和初始化应用

1. SECRET_KEY = "your_secret_key"  # 用于签名 JWT 的密钥
2. ALGORITHM = "HS256"             # 加密算法
3. ACCESS_TOKEN_EXPIRE_MINUTES = 30  # Token 过期时间
5. app = FastAPI()
7. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

复制代码

创建数据模型

1. class Token(BaseModel):
2.     access_token: str
3.     token_type: str
5. class TokenData(BaseModel):
6.     username: str | None = None
8. class User(BaseModel):
9.     username: str
10.     email: str | None = None
11.     full_name: str | None = None
12.     disabled: bool | None = None
14. class UserInDB(User):
15.     hashed_password: str

复制代码

实现辅助函数

生成 JWT Token

1. def create_access_token(data: dict, expires_delta: timedelta | None = None):
2.     to_encode = data.copy()
3.     if expires_delta:
4.         expire = datetime.now(timezone.utc) + expires_delta
5.     else:
6.         expire = datetime.now(timezone.utc) + timedelta(minutes=15)
7.     to_encode.update({"exp": expire})
8.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
9.     return encoded_jwt

复制代码

获取用户数据

1. fake_users_db = {
2.     "testuser": {
3.         "username": "testuser",
4.         "full_name": "Test User",
5.         "email": "testuser@example.com",
6.         "hashed_password": "fakehashedpassword",
7.         "disabled": False,
8.     }
9. }
11. def get_user(db, username: str):
12.     if username in db:
13.         user_dict = db[username]
14.         return UserInDB(**user_dict)

复制代码

验证密码

1. def verify_password(plain_password, hashed_password):
2.     return plain_password == hashed_password

复制代码

获取当前用户

1. async def get_current_user(token: str = Depends(oauth2_scheme)):
2.     credentials_exception = HTTPException(
3.         status_code=status.HTTP_401_UNAUTHORIZED,
4.         detail="Could not validate credentials",
5.         headers={"WWW-Authenticate": "Bearer"},
6.     )
7.     try:
8.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
9.         username: str = payload.get("sub")
10.         if username is None:
11.             raise credentials_exception
12.         token_data = TokenData(username=username)
13.     except jwt.PyJWTError:
14.         raise credentials_exception
15.     user = get_user(fake_users_db, username=token_data.username)
16.     if user is None:
17.         raise credentials_exception
18.     return user
20. async def get_current_active_user(current_user: User = Depends(get_current_user)):
21.     if current_user.disabled:
22.         raise HTTPException(status_code=400, detail="Inactive user")
23.     return current_user

复制代码

用户登录获取 Token

1. @app.post("/token", response_model=Token)
2. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
3.     user = get_user(fake_users_db, form_data.username)
4.     if not user or not verify_password(form_data.password, user.hashed_password):
5.         raise HTTPException(
6.             status_code=status.HTTP_401_UNAUTHORIZED,
7.             detail="Incorrect username or password",
8.             headers={"WWW-Authenticate": "Bearer"},
9.         )
10.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
11.     access_token = create_access_token(
12.         data={"sub": user.username}, expires_delta=access_token_expires
13.     )
14.     return {"access_token": access_token, "token_type": "bearer"}

复制代码

受保护的路由示例

1. @app.get("/users/me/", response_model=User)
2. async def read_users_me(current_user: User = Depends(get_current_active_user)):
3.     return current_user

复制代码

全部代码

1. from fastapi import FastAPI, Depends, HTTPException, status
2. from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
3. from pydantic import BaseModel
4. from datetime import datetime, timedelta, timezone
5. import uvicorn
6. import jwt
7. import os
9. # JWT 相关配置
10. SECRET_KEY = "123456789ashdgjha.slakdv.laksd*as-d/sd3"  # 用于签名 JWT 的密钥（需要妥善保管，实际应用中应存储在环境变量或配置文件中）
11. ALGORITHM = "HS256"  # 使用的加密算法
12. ACCESS_TOKEN_EXPIRE_MINUTES = 30  # Token 的有效时间，以分钟为单位
14. app = FastAPI()  # 创建 FastAPI 应用实例
16. # OAuth2PasswordBearer 实例，用于依赖项
17. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
19. # 模拟的用户数据库，通常在实际应用中应从数据库中获取用户信息
20. fake_users_db = {
21.     "testuser": {
22.         "username": "testuser",
23.         "full_name": "Test User",
24.         "email": "testuser@example.com",
25.         "hashed_password": "fakehashedpassword",  # 在实际应用中，存储经过哈希处理的密码
26.         "disabled": False,  # 用户是否被禁用
27.     }
28. }
30. # Pydantic 模型，用于定义请求和响应的数据结构
31. class Token(BaseModel):
32.     access_token: str  # Token 字符串
33.     token_type: str  # Token 类型（一般为 "bearer"）
35. class TokenData(BaseModel):
36.     username: str | None = None  # 从 Token 中提取的用户名
38. class User(BaseModel):
39.     username: str  # 用户名
40.     email: str | None = None  # 邮箱地址，可选
41.     full_name: str | None = None  # 用户全名，可选
42.     disabled: bool | None = None  # 用户是否被禁用，可选
44. class UserInDB(User):
45.     hashed_password: str  # 存储在数据库中的哈希密码
47. # 生成 JWT Token 的函数
48. def create_access_token(data: dict, expires_delta: timedelta | None = None):
49.     """
50.     生成 JWT Token。
52.     参数：
53.     - data (dict): 要编码到 JWT 中的数据。
54.     - expires_delta (timedelta, 可选): Token 的过期时间。
56.     返回：
57.     - str: 编码后的 JWT 字符串。
58.     """
59.     to_encode = data.copy()  # 创建副本，以避免修改原始数据
60.     if expires_delta:
61.         expire = datetime.now(timezone.utc) + expires_delta
62.     else:
63.         expire = datetime.now(timezone.utc) + timedelta(minutes=15)
64.     to_encode.update({"exp": expire})  # 添加过期时间到 JWT 数据中
65.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)  # 生成 JWT
66.     return encoded_jwt
68. # 从假数据库获取用户信息
69. def get_user(db, username: str):
70.     """
71.     从数据库中获取用户信息。
73.     参数：
74.     - db (dict): 用户数据库（在本例中为假数据）。
75.     - username (str): 用户名。
77.     返回：
78.     - UserInDB | None: 返回匹配的用户信息，如果不存在则返回 None。
79.     """
80.     if username in db:
81.         user_dict = db[username]
82.         return UserInDB(**user_dict)
84. # 验证用户密码
85. def verify_password(plain_password, hashed_password):
86.     """
87.     验证用户密码。
89.     参数：
90.     - plain_password (str): 用户输入的明文密码。
91.     - hashed_password (str): 存储在数据库中的哈希密码。
93.     返回：
94.     - bool: 密码匹配返回 True，否则返回 False。
95.     """
96.     return plain_password == hashed_password  # 在实际应用中，这里应该使用哈希函数进行比较
98. # 验证 Token 并获取当前用户
99. async def get_current_user(token: str = Depends(oauth2_scheme)):
100.     """
101.     从 JWT Token 中提取用户信息，并验证 Token 的合法性。
103.     参数：
104.     - token (str): JWT Token。
106.     返回：
107.     - User: 返回当前用户信息。
109.     抛出：
110.     - HTTPException: 当 Token 无效或用户不存在时抛出 401 错误。
111.     """
112.     credentials_exception = HTTPException(
113.         status_code=status.HTTP_401_UNAUTHORIZED,
114.         detail="Could not validate credentials",
115.         headers={"WWW-Authenticate": "Bearer"},
116.     )
117.     try:
118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])  # 解码 JWT
119.         username: str = payload.get("sub")  # 获取 JWT 中的用户名
120.         if username is None:
121.             raise credentials_exception
122.         token_data = TokenData(username=username)
123.     except jwt.PyJWTError:
124.         raise credentials_exception
125.     user = get_user(fake_users_db, username=token_data.username)  # 获取用户信息
126.     if user is None:
127.         raise credentials_exception
128.     return user
130. # 验证用户是否被禁用
131. async def get_current_active_user(current_user: User = Depends(get_current_user)):
132.     """
133.     验证当前用户是否被禁用。
135.     参数：
136.     - current_user (User): 当前用户信息。
138.     返回：
139.     - User: 如果用户未被禁用，返回用户信息。
141.     抛出：
142.     - HTTPException: 当用户被禁用时抛出 400 错误。
143.     """
144.     if current_user.disabled:
145.         raise HTTPException(status_code=400, detail="Inactive user")
146.     return current_user
148. # 用户登录获取 Token
149. @app.post("/token", response_model=Token)
150. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
151.     """
152.     用户登录接口，用于获取 JWT Token。
154.     参数：
155.     - form_data (OAuth2PasswordRequestForm): 包含用户名和密码的表单数据。
157.     返回：
158.     - dict: 包含 access_token 和 token_type 的响应数据。
160.     抛出：
161.     - HTTPException: 当用户名或密码错误时抛出 401 错误。
162.     """
163.     user = get_user(fake_users_db, form_data.username)
164.     if not user or not verify_password(form_data.password, user.hashed_password):
165.         raise HTTPException(
166.             status_code=status.HTTP_401_UNAUTHORIZED,
167.             detail="Incorrect username or password",
168.             headers={"WWW-Authenticate": "Bearer"},
169.         )
170.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)  # 设置 Token 过期时间
171.     access_token = create_access_token(
172.         data={"sub": user.username}, expires_delta=access_token_expires
173.     )
174.     return {"access_token": access_token, "token_type": "bearer"}
176. # 受保护的路由示例
177. @app.get("/users/me/", response_model=User)
178. async def read_users_me(current_user: User = Depends(get_current_active_user)):
179.     """
180.     获取当前用户信息的受保护路由。
182.     参数：
183.     - current_user (User): 当前登录的用户信息（通过 JWT 验证）。
185.     返回：
186.     - User: 返回当前用户的信息。
187.     """
188.     return current_user
190. # 运行应用
191. if __name__ == "__main__":
192.     uvicorn.run(
193.         f"{os.path.basename(__file__).split('.')[0]}:app",
194.         host="127.0.0.1",
195.         port=8000,
196.         reload=True,  # 启用自动重载
197.     )

复制代码

测试

获取 Token

访问受保护的路由

token正确

token错误

总结

   JWT（JSON Web Token）是一种用于安全地在各方之间传递信息的开放标准，通常用于用户认证和授权。它将用户信息编码为一个签名的令牌，客户端可以使用该令牌访问受保护的资源。
    在 FastAPI 中，JWT 用于保护 API 路由。用户通过提交用户名和密码获取 JWT，客户端在后续请求中使用该令牌进行身份验证。服务器验证令牌的正当性后，答应用户访问受保护的资源。
    实现 JWT 认证的步调包括安装必要的依靠项、配置 JWT 设置（如密钥和算法）、定义数据模型、实现辅助函数（如生成和验证 JWT 的函数），并设置受保护的 API 路由。
   

• JWT 配置：包括密钥（SECRET_KEY）和加密算法（ALGORITHM），用于生成和验证令牌。
  
• 辅助函数：如生成 JWT、验证密码、获取用户信息等，用于处置处罚认证逻辑。
  
• API 路由：包含登录接口，用于生成令牌，以及受保护的路由，只答应携带有效 JWT 的请求访问。
  

    使用 Postman 或其他工具可以测试 JWT 的生成和验证过程，确保只有颠末身份验证的用户才气访问受保护的 API 路由。
  注意

   示例中密码并没有颠末hash加密，现实应用中要加密的。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。