PyJWT 登录鉴权最佳实践【Refresh Token】

🎉 Json web token (JWT), 根据官网的定义，是为了在网络应用环境间传递声明而执行的一种基于JSON的开放标准（RFC 7519)。该token被设计为紧凑且安全的，特别适用于分布式站点的单点登录（SSO）场景。JWT的声明一般被用来在身份提供者和服务提供者间传递被认证的用户身份信息，以便于从资源服务器获取资源，也可以增加一些额外的其它业务逻辑所必须的声明信息，该token也可直接被用于认证，也可被加密。

本文将针对 JWT 的缺点，来通过 refresh_token 方案做一个最佳解决方案。

Welcome to PyJWT — PyJWT 2.8.0 documentationhttps://pyjwt.readthedocs.io/en/stable/

1. JWT 优缺点

1.1. JWT 优点

跨域和跨平台: JWT 由于基于 JSON，具有很好的跨语言支持，可以在任何 web 环境和多种编程语言之间使用。
自包含: JWT内部包含了所有用户状态信息（claims），从而无需在服务器端存储会话信息，便于分布式系统的扩展。
CSRF保护: 由于 JWT 不依赖于 Cookie，因此它天然地防护了跨站请求伪造（CSRF）攻击。
轻量和可拓展: JWT本身结构紧凑，传输快捷，并且可以存储额外的业务逻辑所需的非敏感信息。
无状态和可扩展性: JWT 的无状态特性简化了服务器设计，不需要额外的存储如 Redis 来管理会话状态，有利于服务的横向扩展。

1.2. JWT 缺点

Token失效问题: 一旦 JWT 被签发，在有效期内将持续有效，直到过期。服务器无法控制 Token 的失效，除非引入黑名单机制，这增加了系统的复杂度。
安全性风险: JWT 如果被截获，由于其自包含的特性，如果不采取额外的保护措施（如HTTPS），就可能被利用。
存储问题: JWT 通常存储在客户端，如果客户端安全措施不到位，可能会导致 Token 的泄露。
性能问题: 由于每次请求都需要传输 JWT，如果 Token 过大则会增加请求负载，此外服务器每次都需要验证 Token 的签名，这可能会引入一定的性能开销。
Token刷新问题: 在长期有效性的 JWT 系统中，如何安全地刷新 Token 也是一个挑战，需要设计合理的 Token 刷新机制。

1.3. refresh_token 解决方案

既然 JWT 依然存在诸多问题，甚至无法满足一些业务上的需求，但是我们依然可以基于 JWT 在实践中进行一些改进，来形成一个折中的方案。前面讲的 Token，都是 Access Token，也就是访问资源接口时所需要的 Token，还有另外一种 Token，Refresh Token，通常情况下，Refresh Token 的有效期会比较长，而 Access Token 的有效期比较短，当 Access Token 由于过期而失效时，使用 Refresh Token 就可以获取到新的 Access Token，如果 Refresh Token 也失效了，用户就只能重新登录了。

所以，在 JWT 的实践中，我决定引入 Refresh Token，将会话管理流程改进如下。

客户端使用用户名+密码登录
- 服务端生成有效时间较短的 Access Token（60 分钟）和有效时间较长的 Refresh Token（24 小时）
- 关于 Refresh Token 只要重新登录，就要刷新 redis 的 Refresh Token，避免突然过期情况。
- Access Token 和 Refresh Token 的失效时间设置建议【见后文】
客户端访问需要认证的接口时（因为有的接口可能会设置为@NoAuth），携带 Access Token。
- 如果 Access Token 没有过期，服务端鉴权后返回给客户端需要的数据。
客户端需要在 Refresh Token 即将过期时（例如，距离过期还有1分钟），自动使用 Refresh Token 去请求新的Access Token。这样可以提高用户体验，避免在用户操作过程中突然需要重新登录。
如果 token 失效，则在需要认证的接口请求中，服务端需返回 401 错误，客户端此时需要重新登录
如果 Refresh Token 没有过期，服务端向客户端下发新的 Access Token，接下来客户端需使用新的 Access Token 访问需要认证的接口。
注销登录：
- Refresh Token 立即从 redis 删除（即失效）
- Access Token 放入黑名单中，在认证接口请求校验过程中，需要同步校验一下

2. JWT 介绍

官网地址：Welcome to PyJWT — PyJWT 2.8.0 documentation

安装：pip install pyjwt

JWT规定的协议的格式 RFC 7515 - JSON Web Signature (JWS)

由三部分组成：头部（Header）、负载（Payload）和签名（Signature）。

2.1. header

声明类型，这里是jwt
声明加密的算法通常直接使用 HMAC SHA256

一种常见的头部是这样的：

{
  'typ': 'JWT',
  'alg': 'HS256'
}

然后再将其进行base64编码。

2.2. payload 负载

JWT（JSON Web Token）的有效负载（payload）通常包含一系列声明（claims），这些声明是关于实体（通常是指用户）和其他数据的声明。有效负载中包含的声明可以分为三种类型：注册的声明、公共的声明和私有的声明。

注册的声明：这些是预先定义的声明，它们不是强制的，但它们的使用是推荐的。它们包括：
- iss (Issuer)：声明 token 的发行者。
- sub (Subject)：声明的主题（通常是用户 ID）。
- aud (Audience)：声明的受众。
- exp (Expiration Time)：token 的过期时间。
- nbf (Not Before)：在此时间之前，token 不可用。
- iat (Issued At)：token 的发行时间。
- jti (JWT ID)：JWT 的唯一身份标识，主要用来作为一次性 token，从而回避重放攻击。
公共的声明：这些可以被任何人使用，并且不是由 JWT 标准预先定义的。当使用公共的声明时，应该在 IANA JSON Web Token Claims 注册表中进行定义，或者包含一个包含冲突避免命名空间的 URI。
私有的声明：这些是由发送者和接收者共同定义的声明，不是注册或公共的声明。这些声明是为了在双方之间传递信息而建立的，并且不是 JWT 标准定义的，也不在 IANA 注册表中注册。

一个常见的payload是这样的：


{
  "sub": "1234567890",
  "name": "John Doe",
  "admin": true,
  "iat": 1516239022
}

在这个例子中，sub 是一个注册的声明，表示用户的唯一标识符。name 和 admin 是私有的声明，它们是关于用户的具体信息。iat 是一个注册的声明，表示 token 的发行时间。

2.3. signature

signature 存储了序列化的 secreate key 和 salt key。这个部分需要 base64 加密后的 header 和 base64 加密后的 payload 使用.连接组成的字符串，然后通过 header 中声明的加密方式进行加盐 secret 组合加密，然后就构成了jwt的第三部分。

Access Token 和 Refresh Token 的失效时间设置取决于多种因素，包括应用的安全需求、用户体验和常见的行业实践。以下是一些通用的指导原则和建议：

3. Python 实现 demo

以下通过一个简单的 Python demo 程序来实现上述的认证方案。这个 demo 程序会包括几个基本的文件：

main.py - 主应用程序，设置 FpastAPI web 服务。
auth.py - 处理认证逻辑，例如生成、刷新、验证 Token 以及黑名单处理。
blacklist.py - 处理黑名单的逻辑。
user.py - 用户模型和用户管理。

以下是每个文件的简单示例实现：

3.1. main.py

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from auth import (
authenticate_user,
create_access_token,
create_refresh_token,
verify_token,
ALGORITHM,
SECRET_KEY
)
from blacklist import add_token_to_blacklist
import aioredis
from jose import jwt

app = FastAPI()

# 配置 Redis 连接
REDIS_URL = "redis://localhost:6379"
redis = aioredis.from_url(REDIS_URL, decode_responses=True)

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

@app.post("/login")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    username = form_data.username
    password = form_data.password
    # 这里应该验证用户名和密码，这里只是示例，所以我们假设用户验证成功
    if authenticate_user(username, password):
        access_token = create_access_token(data={"sub": username})
        refresh_token = create_refresh_token(data={"sub": username})
        # 将 refresh token 存储到 Redis 中
        await redis.setex(username, 3600*24, refresh_token)
        return {"access_token": access_token, "token_type": "bearer", "refresh_token": refresh_token}
    else:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )

@app.post("/logout")
async def logout(token: str = Depends(oauth2_scheme)):
    add_token_to_blacklist(token)
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    username: str = payload.get("sub")
    if username is not None:
        # 从 Redis 中删除 refresh token
        await redis.delete(username)
    return {"message": "Logout successful"}

# ... 其他的 endpoint 和 helper 函数 ...

3.2. auth.py

from jose import jwt, JWTError
from datetime import datetime, timedelta
from fastapi import HTTPException, status

SECRET_KEY = "your_super_secret_key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_HOURS = 24

def authenticate_user(username: str, password: str):
    # 这里应该有用户验证的逻辑，这里只是示例
    return True  # 假设用户验证成功

def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(hours=REFRESH_TOKEN_EXPIRE_HOURS)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def verify_token(token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        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
        return username
    except JWTError:
        raise credentials_exception

3.3. blacklist.py

BLACKLIST = set()

def add_token_to_blacklist(token: str):
    BLACKLIST.add(token)

def is_token_in_blacklist(token: str):
    return token in BLACKLIST

请注意，上述代码是一个简化的示例，没有实现用户验证和完整的错误处理。在生产环境中，您应该使用安全的方式处理密码验证，并且使用更复杂的认证系统和错误处理机制。此外，您需要确保 SECRET_KEY 被保密，并且是随机生成的，以及在生产环境中，Redis 应该被相应地配置和保护。为了运行这个 FastAPI 应用程序，您需要在您的项目目录下创建上述的 main.py, auth.py, 和 blacklist.py 文件，并将上述代码粘贴到相应的文件中。然后，您可以使用以下命令来启动服务器：uvicorn main:app --reload使用 --reload 参数会使服务器在代码改变时自动重启，这对于开发过程很有帮助，但不应该在生产环境中使用。

3.4. NoAuth 跳过 token 校验

我们在实际业务开发中，有的接口可能不需要 token 校验机制，或者后端开发调试接口时，就需要跳过 token 校验。

而在 FastAPI 中，可以通过依赖注入系统灵活地控制哪些接口需要进行Token校验，哪些接口不需要。虽然 FastAPI 没有直接提供类似于 @NoAuth 注解的功能，但我们可以通过定义不同的依赖项来实现类似的效果。

【其实，不写这个也行，就是 Depends() 为空即可】

以下是一个简单的示例，展示如何在某些接口上跳过Token校验：

定义一个可选的 token 校验依赖项

首先，我们可以定义一个依赖项函数，该函数尝试校验Token，但如果没有提供Token，它也会允许请求通过。这可以通过捕获特定的异常来实现，或者通过检查Token是否存在。

from fastapi import Depends, HTTPException, Security
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

async def no_auth(token: str = Security(oauth2_scheme, auto_error=False)):
    if not token:
        return None  # 如果没有提供Token，直接返回 None，表示跳过校验
    try:
        # 这里是Token校验逻辑，如果校验失败，抛出异常
        return verify_token(token)
    except JWTError:
        raise credentials_exception

在需要跳过Token校验的接口上使用这个依赖项Depends(no_auth)：

from fastapi import FastAPI, Depends

app = FastAPI()

@app.get("/xxx")
async def no_auth_required(user=Depends(no_auth)):
    return {"message": "No auth required for this endpoint."}

在这个例子中，/xxx 路由不需要Token即可访问。如果提供了Token，它会尝试进行校验，并且如果校验成功，user变量将包含用户信息（在这个例子中是用户名）。这种方法提供了灵活性，允许您根据需要在每个路由函数中决定是否进行Token校验。通过使用不同的依赖项，您可以轻松地为不同的接口设置不同的认证和授权策略。

4. token 失效时间设置建议

4.1. Access Token 失效时间

较短的有效期：Access Token 应当有一个较短的有效期，通常在几分钟到几小时之间。短期有效可以减少如果Token被泄露后可能造成的损害。
推荐时长：一个常见的实践是将 Access Token 的有效期设置为 15 分钟到 1 小时。这样的设置能够在保障安全的同时，减少需要用户重新认证的次数。

4.2. Refresh Token 失效时间

较长的有效期：Refresh Token 的有效期通常比 Access Token 长。这是因为 Refresh Token 用于在 Access Token 过期后获取新的 Access Token，而不需要用户重新登录。
推荐时长：Refresh Token 的有效期可以设置得更长，一般从几天到几个月不等。一些应用可能会根据用户的活跃程度来调整有效期，如在用户活跃时自动续期。
固定过期时间：考虑到安全性，Refresh Token 最好有一个固定的过期时间，并且不应该通过每次使用就延长有效期的方式来续期。
安全考量：如果 Refresh Token 被泄露，攻击者可以使用它来获取新的 Access Token。因此，即使 Refresh Token 有较长的有效期，也需要确保其他安全措施得到执行，比如限制使用特定设备或IP地址、检测异常使用模式、及时注销机制等。

4.3. 实际设置

依据实际应用场景：您的具体设置应当根据您的应用场景和安全要求来决定。例如，对于高安全性要求的应用（如金融或医疗应用），您可能希望设置较短的有效期。而对于用户体验非常重要的应用，则可能需要较长的有效期以减少用户重新认证的次数。
遵守法规和标准：某些行业可能有关于认证Token有效期的特定法规和标准，确保您的设置符合这些要求。

总的来说，一个合理的起点可能是设置 Access Token 的有效期为 15 分钟到 1 小时，Refresh Token 的有效期为 7 天到 30 天。然后根据您对安全性和用户体验的权衡，以及用户和业务的实际需求，对这些值进行调整。