【进阶】FastAPI中的用户认证与授权

# 1. FastAPI概述**

FastAPI是一个现代、高性能的Python Web框架，专为构建API和微服务而设计。它提供了一系列强大的功能，包括：

- **异步支持：**FastAPI基于ASGI（异步服务器网关接口），支持异步编程，可实现高并发和低延迟。
- **数据验证：**FastAPI集成了Pydantic数据验证库，可轻松验证请求和响应数据，确保数据完整性和一致性。
- **依赖注入：**FastAPI支持依赖注入，允许在应用程序中轻松创建和管理依赖关系，提高代码的可测试性和可维护性。
- **OpenAPI文档：**FastAPI自动生成OpenAPI文档，用于描述API的端点、参数和响应，方便客户端集成和测试。

# 2. FastAPI用户认证

### 2.1 用户认证机制

用户认证是验证用户身份的过程，以确保只有授权用户才能访问受保护的资源。FastAPI提供了多种认证机制，包括：

- **基于令牌的认证：**使用JSON Web令牌（JWT）或其他令牌类型来验证用户身份。
- **会话认证：**使用会话ID或cookie来跟踪用户会话。
- **基本认证：**使用用户名和密码进行身份验证。
- **OAuth2认证：**使用OAuth2协议进行身份验证，允许用户通过第三方服务（如Google或Facebook）登录。

### 2.2 JWT令牌的使用

JWT（JSON Web令牌）是一种轻量级、紧凑的令牌，用于在两个 parties 之间安全地传输信息。JWT令牌由三部分组成：

- **头部：**包含令牌类型和签名算法等元数据。
- **有效载荷：**包含有关用户的声明，例如用户名、电子邮件和角色。
- **签名：**使用私钥对令牌进行签名，以确保其完整性和真实性。

#### 2.2.1 JWT令牌的生成和验证

要生成JWT令牌，可以使用以下代码：

import jwt

payload = {
    "username": "john",
    "email": "john@example.com",
    "role": "admin"
}

secret_key = "my_secret_key"
token = jwt.encode(payload, secret_key, algorithm="HS256")

要验证JWT令牌，可以使用以下代码：

import jwt

secret_key = "my_secret_key"
token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6ImpvaG4iLCJlbWFpbCI6ImpvaG5AZXhhbXBsZS5jb20iLCJyb2xlIjoiYWRtaW4ifQ.23123123123123"

try:
    decoded_token = jwt.decode(token, secret_key, algorithms=["HS256"])
    print(decoded_token)
except jwt.DecodeError:
    print("Invalid token")

#### 2.2.2 JWT令牌的过期时间和刷新

JWT令牌可以设置过期时间，以确保其在特定时间后失效。可以使用`exp`声明来指定过期时间，单位为自纪元时间（Epoch Time）以来的秒数。

payload = {
    "username": "john",
    "email": "john@example.com",
    "role": "admin",
    "exp": time.time() + 60 * 60  # 1 hour from now
}

要刷新JWT令牌，可以生成一个新的令牌并替换旧令牌。

### 2.3 基于令牌的认证中间件

FastAPI提供了`HTTPBearerAuthorization`中间件，用于基于令牌进行认证。该中间件从请求头中提取令牌，并将其传递给认证函数进行验证。

#### 2.3.1 中间件的实现和使用

from fastapi import FastAPI, HTTPBearer
from fastapi.security import HTTPAuthorizationCredentials

app = FastAPI()

# 创建HTTPBearerAuthorization中间件
auth_middleware = HTTPBearer()

# 定义认证函数
async def verify_token(credentials: HTTPAuthorizationCredentials):
    try:
        decoded_token = jwt.decode(credentials.credentials, "my_secret_key", algorithms=["HS256"])
        return decoded_token
    except jwt.DecodeError:
        return None

# 将中间件添加到FastAPI应用中
app.add_middleware(
    auth_middleware,
    verify=verify_token,
    path="/protected-route"
)

#### 2.3.2 中间件的自定义和扩展

`HTTPBearerAuthorization`中间件可以进行自定义和扩展，以满足特定的认证需求。例如，可以自定义令牌验证函数以支持不同的令牌类型或验证算法。

# 3. FastAPI用户授权

### 3.1 权限管理模型

权限管理模型是定义和管理用户权限的框架。FastAPI支持多种权限管理模型，包括：

- **基于角色的授权 (RBAC)**：用户被分配角色，每个角色具有特定的权限。
- **基于作用域的授权 (SBAC)**：用户被分配作用域，每个作用域定义了用户可以访问的资源。
- **基于属性的授权 (ABAC)**：用户被分配属性，例如部门或职称，权限基于这些属性进行评估。

### 3.2 基于角色的授权

#### 3.2.1 角色的定义和管理

角色是用户组，具有特定的权限集。在FastAPI中，可以使用`@role_required`装饰器来保护端点，仅允许具有指定角色的用户访问。

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer
from jose import jwt
from pydantic import BaseModel

class User(BaseModel):
    username: str
    role: str

security = HTTPBearer()

def get_current_user(token: str = Depends(security)):
    try:
        payload = jwt.decode(token, "secret", algorithms=["HS256"])
        user = User(**payload)
        return user
    except Exception:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
        )

@app.get("/protected")
async def protected(user: User = Depends(get_current_user)):
    if user.role != "admin":
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Insufficient permissions",
        )
    return {"message": "Hello, admin!"}

#### 3.2.2 基于角色的访问控制

基于角色的访问控制 (RBAC) 是一种授权机制，它允许管理员将权限分配给角色，然后将角色分配给用户。这使得管理权限变得更加容易，因为管理员只需要更新角色的权限，而不是逐个更新每个用户的权限。

graph LR
subgraph RBAC
    A[Admin] -