ToB企服应用市场:ToB评测及商务社交产业平台

标题: PyJWT 和 python-jose 在处理JWT令牌处理的时候的差异和详细使用 [打印本页]

作者: 卖不甜枣    时间: 2024-8-27 09:35
标题: PyJWT 和 python-jose 在处理JWT令牌处理的时候的差异和详细使用
PyJWT 和 python-jose 是两个用于处理 JSON Web Tokens (JWT) 的 Python 库。它们都有助于天生、解码、验证和管理 JWT,但它们在功能范围和设计哲学上有一些重要的区别。本篇先容它们之间的一些差异,以及在项目中使用FastAPI+ python-jose 来处理访问令牌的天生以及一些例子代码供参考。
1、PyJWT 和 python-jose的差异

PyJWT

PyJWT 是一个专门处理 JWT 的 Python 库,它旨在简化 JWT 的创建和验证。
特点:
主要用法示例:
  1. import jwt
  2. import datetime
  3. # 创建一个JWT
  4. payload = {
  5.     "user_id": 123,
  6.     "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
  7. }
  8. secret = 'your-secret-key'
  9. token = jwt.encode(payload, secret, algorithm='HS256')
  10. # 解码JWT
  11. decoded = jwt.decode(token, secret, algorithms=['HS256'])
  12. print(decoded)
复制代码
python-jose

python-jose 是一个更广泛的加密库,它不但支持 JWT,还支持多种 JOSE (JSON Object Signing and Encryption) 标准,包罗 JWS (JSON Web Signature)、JWE (JSON Web Encryption)、JWK (JSON Web Key)、JWA (JSON Web Algorithms) 等。
特点:
主要用法示例:
  1. from jose import jwt
  2. from jose.exceptions import JWTError
  3. # 创建一个JWT
  4. payload = {
  5.     "user_id": 123,
  6.     "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
  7. }
  8. secret = 'your-secret-key'
  9. token = jwt.encode(payload, secret, algorithm='HS256')
  10. # 解码JWT
  11. try:
  12.     decoded = jwt.decode(token, secret, algorithms=['HS256'])
  13.     print(decoded)
  14. except JWTError as e:
  15.     print(f"Token is invalid: {e}")
复制代码
差异总结

 
2、使用 python-jose 处理 JWT 

在使用 python-jose 处理 JWT 时,捕获和处理非常是一个重要的环节。
1) 安装 python-jose

起首,确保你已经安装了 python-jose:
  1. pip install python-jose
复制代码
2)使用 python-jose 的 JWT 模块

以下是一个使用 python-jose 的 JWT 处理的示例,包罗怎样捕获非常:
  1. from jose import jwt, JWTError
  2. from jose.exceptions import ExpiredSignatureError
  3. # 定义密钥和有效负载
  4. secret = 'your-secret-key'
  5. payload = {
  6.     "user_id": 123,
  7.     "exp": datetime.datetime.utcnow() + datetime.timedelta(hours=1)
  8. }
  9. # 生成 JWT
  10. token = jwt.encode(payload, secret, algorithm='HS256')
  11. # 解码 JWT 并处理可能的异常
  12. try:
  13.     decoded = jwt.decode(token, secret, algorithms=['HS256'])
  14.     print(decoded)
  15. except ExpiredSignatureError:
  16.     print("Token has expired")
  17. except JWTError:
  18.     print("Token is invalid")
复制代码
在处理 python-jose 的 JWT 时,正确地捕获和处理非常是关键。确保你的环境和工具能够正确识别非常类型,将有助于你更好地管理 JWT 错误。
假如我们需要再JWT的playload里面承载更多的信息,可以再claim中声明键值即可,你可以使用 python-jose 和 FastAPI 来实现 JWT 令牌天生操纵。如下所示代码。
使用 FastAPI 和 python-jose 天生 JWT 令牌:
  1. from fastapi import FastAPI, Depends, HTTPException, status, Request
  2. from fastapi.responses import JSONResponse
  3. from jose import JWTError, jwt
  4. from datetime import datetime, timedelta
  5. app = FastAPI()
  6. # 配置项,通常从配置文件中加载
  7. JWT_SECRET_KEY = 'your_jwt_secret_key'
  8. JWT_ISSUER = 'your_issuer'
  9. JWT_AUDIENCE = 'your_audience'
  10. JWT_EXPIRED_DAYS = 7
  11. ALGORITHM = 'HS256'
  12. def generate_token(user_info: dict, role_type: str):
  13.     # 获取IP地址
  14.     ip = user_info.get('ip', '')
  15.     # 定义声明
  16.     claims = {
  17.         'id': user_info['id'],
  18.         'email': user_info['email'],
  19.         'name': user_info['name'],
  20.         'nickname': user_info.get('nickname', ''),
  21.         'phone_number': user_info.get('mobile_phone', ''),
  22.         'gender': user_info.get('gender', ''),
  23.         'full_name': user_info.get('full_name', ''),
  24.         'company_id': user_info.get('company_id', ''),
  25.         'company_name': user_info.get('company_name', ''),
  26.         'dept_id': user_info.get('dept_id', ''),
  27.         'dept_name': user_info.get('dept_name', ''),
  28.         'role_type': role_type,
  29.         'ip': ip,
  30.         'mac_addr': '',  # 无法获得Mac地址
  31.         'channel': ''
  32.     }
  33.     # 定义token过期时间
  34.     expiration = datetime.utcnow() + timedelta(days=JWT_EXPIRED_DAYS)
  35.     # 创建JWT token
  36.     to_encode = {
  37.         **claims,
  38.         'iss': JWT_ISSUER,
  39.         'aud': JWT_AUDIENCE,
  40.         'exp': expiration
  41.     }
  42.     token = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=ALGORITHM)
  43.     return token
复制代码
  1. @app.post('/token')
  2. async def get_token(request: Request):
  3.     # 模拟的用户信息
  4.     user_info = {
  5.         'id': 123,
  6.         'email': 'user@example.com',
  7.         'name': 'John Doe',
  8.         'nickname': 'Johnny',
  9.         'mobile_phone': '123-456-7890',
  10.         'gender': 'Male',
  11.         'full_name': 'Johnathan Doe',
  12.         'company_id': 'ABC123',
  13.         'company_name': 'ABC Corp',
  14.         'dept_id': 'Dept001',
  15.         'dept_name': 'IT',
  16.         'ip': request.client.host
  17.     }
  18.     role_type = 'Admin'
  19.     token = generate_token(user_info, role_type)
  20.    
  21.     headers = {
  22.         'access-token': token,
  23.         'Authorization': f'Bearer {token}'
  24.     }
  25.     return JSONResponse(content={'token': token}, headers=headers)
  26. if __name__ == "__main__":
  27.     import uvicorn
  28.     uvicorn.run(app, host="127.0.0.1", port=8000)
复制代码
解释

 
3、在api中怎样实现AllowAnonymous和验证授权

在 Python 的 FastAPI 框架中,你可以通过以下方式实现雷同于 ASP.NET 中的 AllowAnonymous 和授权验证功能。
1)实现 JWT 授权验证中间件

起首,你需要一个依赖项来检查请求中是否包含有效的 JWT 令牌。假如令牌无效或缺失,依赖项将拒绝请求。通过这种方式,只有标记为“允许匿名”的路由才会跳过验证。
2)安装依赖

确保安装了 python-jose 用于处理 JWT,以及 fastapi:
3) 创建授权依赖

你可以创建一个名为 get_current_user 的依赖项,用于验证 JWT 令牌并提取用户信息。假如没有提供或验证失败,抛出 HTTPException。
  1. from fastapi import Depends, HTTPException, status
  2. from jose import JWTError, jwt
  3. from typing import Optional
  4. JWT_SECRET_KEY = 'your_jwt_secret_key'
  5. ALGORITHM = 'HS256'
  6. def get_current_user(token: str = Depends(oauth2_scheme)):
  7.     try:
  8.         payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM])
  9.         user_id: str = payload.get("id")
  10.         if user_id is None:
  11.             raise HTTPException(
  12.                 status_code=status.HTTP_401_UNAUTHORIZED,
  13.                 detail="Could not validate credentials",
  14.                 headers={"WWW-Authenticate": "Bearer"},
  15.             )
  16.         return payload
  17.     except JWTError:
  18.         raise HTTPException(
  19.             status_code=status.HTTP_401_UNAUTHORIZED,
  20.             detail="Could not validate credentials",
  21.             headers={"WWW-Authenticate": "Bearer"},
  22.         )
复制代码
get_current_user:用于解析和验证 JWT。假如令牌无效或缺失,会抛出 HTTPException,返回 401 状态码。
详细使用的时候,我们可能把用户信息缓存在Redis里面提高处理效率。
4)实现 AllowAnonymous 功能

在 FastAPI 中,你可以通过 Depends 来实现条件授权。对于需要授权的路由,只需将 get_current_user 作为依赖传递给路由函数。而无需授权的路由,可以直接定义。
  1. from fastapi import FastAPI, Depends, HTTPException, status
  2. from fastapi.security import OAuth2PasswordBearer
  3. app = FastAPI()
  4. oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
  5. # 允许匿名访问的路由
  6. @app.get("/public")
  7. def read_public_data():
  8.     return {"message": "This is a public endpoint"}
  9. # 需要授权的路由
  10. @app.get("/protected")
  11. def read_protected_data(current_user: dict =<strong> Depends(get_current_user)</strong>):
  12.     return {"message": f"Hello, {current_user['name']}"}
  13. # 模拟登录生成 JWT 令牌的路由
  14. @app.post("/token")
  15. def login():
  16.     # 这里省略了身份验证过程,只是直接生成一个 JWT 令牌
  17.     token = jwt.encode({"id": 1, "name": "John Doe"}, JWT_SECRET_KEY, algorithm=ALGORITHM)
  18.     return {"access_token": token, "token_type": "bearer"}
复制代码
Depends(get_current_user):在需要保护的路由中,通过依赖项注入 get_current_user,确保只有通过身份验证的用户才能访问。
在 FastAPI 中,建议通过依赖项注入(Depends)来获取当前用户的信息,而不是直接访问 request.user。这种方式更加灵活而且与 FastAPI 的设计哲学更同等。
在详细项目中,我们为了方便,每每通过中间件的方式进行定义和处理授权的过程。

通过authentiate函数处理验证用户令牌的有效性。

 我们一般再main.py入口中加入中间件的处理即可。
  1.     # JWT auth, required
  2.     app.add_middleware(
  3.         AuthenticationMiddleware,
  4.         backend=JwtAuthMiddleware(),
  5.         on_error=JwtAuthMiddleware.auth_exception_handler,
  6.     )
复制代码
 
4、一些错误处理

当你在解码 JWT 时遇到 "Invalid audience" 错误,通常意味着在天生或解码 JWT 时,aud (audience) 声明没有正确设置或验证。以下是办理这个问题的步调和阐明:
1) 明白 aud (Audience) 声明

设置和检查 aud 声明,天生 Token 时设置 aud
 
假如你盼望 JWT 包含 aud 声明,可以在天生 token 时传递它。例如:
  1. to_encode = {
  2.     "sub": "user_id",
  3.     "aud": "your_audience",  # 设置aud声明
  4.     "exp": datetime.utcnow() + timedelta(minutes=30)
  5. }
  6. token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)
复制代码
解码 Token 时验证 aud
在解码 JWT 时,指定 audience 参数以匹配天生时的 aud:
  1. try:
  2.     payload = jwt.decode(
  3.         token,
  4.         settings.TOKEN_SECRET_KEY,
  5.         algorithms=[settings.TOKEN_ALGORITHM],
  6.         audience="your_audience"  # 验证aud声明
  7.     )
  8.     print(f"Decoded payload: {payload}")
  9. except JWTError as e:
  10.     print(f"Token decode failed: {e}")
复制代码
2)Token decode failed: Subject must be a string.

"Token decode failed: Subject must be a string" 错误通常是由于 sub (subject) 声明的值不是字符串引起的。sub 是 JWT 中常用的一个声明,用来标识 token 的主体,好比用户 ID 或用户名。JWT 标准要求 sub 的值必须是字符串。
检查 sub 声明的值
起首,确保在天生 token 时,sub 声明的值是一个字符串:
  1. to_encode = {
  2.     "sub": str(user_id),  # 确保 user_id 是字符串
  3.     "aud": "your_audience",  # 其他字段
  4.     "exp": datetime.utcnow() + timedelta(minutes=30)
  5. }
  6. token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)
复制代码
假如 sub 的值是一个非字符串类型(如整数或其他对象),请将其转换为字符串:
  1. user_id = 123  # 假设 user_id 是一个整数
  2. to_encode = {
  3.     "sub": str(user_id),  # 将 user_id 转换为字符串
  4.     "aud": "your_audience",
  5.     "exp": datetime.utcnow() + timedelta(minutes=30)
  6. }
  7. token = jwt.encode(to_encode, settings.TOKEN_SECRET_KEY, algorithm=settings.TOKEN_ALGORITHM)
复制代码
 
3)schema怎样移除敏感字段

在处理数据模型和模式(schema)时,特别是当涉及到敏感信息(如暗码、身份信息等)时,偶然需要从输出或序列化结果中移除这些敏感字段。根据你使用的库或框架,处理敏感字段的方法会有所不同。以下是一些常见的方法来移除敏感字段:
使用 Pydantic 的 exclude 参数
假如你在使用 Pydantic(例如在 FastAPI 中),你可以使用模型的 dict 方法的 exclude 参数来清除敏感字段。
  1. from pydantic import BaseModel
  2. class User(BaseModel):
  3.     username: str
  4.     email: str
  5.     password: str  # 敏感字段
  6. user = User(username="user1", email="user1@example.com", password="secret")
  7. # 创建一个不包含敏感字段的字典
  8. user_dict = user.dict(exclude={"password"})
  9. print(user_dict)
复制代码
使用 SQLAlchemy 的 __mapper_args__
假如你使用 SQLAlchemy 而且盼望从序列化结果中清除敏感字段,可以使用模型的 __mapper_args__ 进行配置。例如:
  1. from sqlalchemy import Column, String
  2. from sqlalchemy.ext.declarative import declarative_base
  3. Base = declarative_base()
  4. class User(Base):
  5.     __tablename__ = 'users'
  6.     id = Column(String, primary_key=True)
  7.     username = Column(String)
  8.     email = Column(String)
  9.     password = Column(String)  # 敏感字段
  10.     def to_dict(self):
  11.         # 移除敏感字段
  12.         return {c.name: getattr(self, c.name) for c in self.__table__.columns if c.name != 'password'}
  13. user = User(id="1", username="user1", email="user1@example.com", password="secret")
  14. print(user.to_dict())
复制代码
自定义序列化方法
假如你使用自定义模型或类,而且没有使用特定的库,你可以实现自定义序列化方法来清除敏感字段。
  1. class User:
  2.     def __init__(self, username, email, password):
  3.         self.username = username
  4.         self.email = email
  5.         self.password = password  # 敏感字段
  6.     def to_dict(self):
  7.         # 移除敏感字段
  8.         return {
  9.             "username": self.username,
  10.             "email": self.email
  11.         }
  12. user = User(username="user1", email="user1@example.com", password="secret")
  13. print(user.to_dict())
复制代码
 

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




欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/) Powered by Discuz! X3.4