如何在FastAPI中巧妙玩转数据脱敏，让敏感信息安全无忧？-博客园

如何在FastAPI中巧妙玩转数据脱敏，让敏感信息安全无忧？

2025-07-02 00:29:10 发布 102 浏览

页面报错/反馈

已收藏点赞

扫描二维码
关注或者微信搜一搜：编程智域前端至全栈交流与成长

发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/

一、Pydantic模型敏感字段标记

1.1 基础字段标记方法

通过Field函数的description参数声明敏感字段：

from pydantic import BaseModel, Field


class User(BaseModel):
    username: str
    password: str = Field(..., description="敏感字段-用户密码")
    phone: str = Field(..., example="138****7890")

1.2 高级安全类型应用

使用SecretStr类型自动隐藏敏感数据：

from pydantic import SecretStr


class SafeUser(BaseModel):
    api_key: SecretStr
    db_password: SecretStr


# 输出时将自动转为********
print(SafeUser(api_key="sk-1234", db_password="db@123").json())
# 输出：{"api_key": "**********", "db_password": "**********"}

1.3 模型配置示例

class SensitiveModel(BaseModel):
    class Config:
        json_encoders = {
            SecretStr: lambda v: "*******" if v else None
        }
        exclude_fields = ['password']

流程图说明：

graph TD A[用户请求] --> B[模型验证] B --> C[敏感字段标记] C --> D[数据持久化]

二、响应数据动态脱敏策略

2.1 响应模型过滤

class UserResponse(BaseModel):
    username: str
    email: str


@app.get("/users/me", response_model=UserResponse)
async def read_user_me():
    return current_user.dict()

2.2 动态字段排除

from fastapi import Depends


def mask_sensitive_data(user: User):
    return user.dict(exclude={"password", "ssn"})


@app.get("/users/{id}")
async def get_user(data: dict = Depends(mask_sensitive_data)):
    return data

2.3 条件脱敏实现

from typing import Optional


class DynamicResponse(BaseModel):
    name: str
    phone: Optional[str]
    email: Optional[str]

    @classmethod
    def create_with_mask(cls, user: User, is_admin: bool):
        fields = user.dict()
        if not is_admin:
            fields.update({"phone": "138****7890", "email": "***@example.com"})
        return cls(**fields)

流程图说明：

graph TD A[原始数据] --> B{权限判断} B -->|通过| C[字段替换] B -->|拒绝| D[返回错误] C --> E[响应输出] D --> E

三、第三方加密服务集成（Vault）

3.1 Vault配置示例

import hvac

vault_client = hvac.Client(
    url="http://vault:8200",
    token="s.4zNq3Z8gKj9R6tY1"
)


def encrypt_data(data: str) -> str:
    return vault_client.secrets.transit.encrypt_data(
        name="fastapi-key",
        plaintext=data.encode()
    )["data"]["ciphertext"]

3.2 集成到数据模型

from pydantic import validator


class EncryptedUser(User):
    @validator('password', pre=True)
    def encrypt_password(cls, v):
        return encrypt_data(v)

3.3 完整工作流程

graph TD A[客户端请求] --> B[参数验证] B --> C{敏感字段判断} C -->|是| D[Vault加密] C -->|否| E[直接存储] D --> F[数据库保存] E --> F

课后Quiz

当需要在前端显示用户手机号时，应该使用哪种脱敏方式？
A) 完全显示
B) 中间四位星号
C) 全部加密
D) 随机替换

答案：B
答案解析：根据PCI DSS规范要求，敏感信息需要部分隐藏但保持可识别性
以下哪项是Vault的核心功能？
A) 自动生成API文档
B) 动态密钥管理
C) 请求速率限制
D) 数据库迁移

答案：B
答案解析：Vault提供加密即服务、密钥轮换等安全功能

常见报错解决方案

问题1：422 Validation Error

现象：
{"detail":[{"loc":["body","password"],"msg":"field required","type":"value_error.missing"}]}

解决方法：

检查请求体是否包含必填字段
验证字段类型是否符合模型定义
使用try-except块捕获ValidationError：

from fastapi import HTTPException
from pydantic import ValidationError


@app.post("/users")
async def create_user(data: dict):
    try:
        return User(**data)
    except ValidationError as e:
        raise HTTPException(422, detail=e.errors())

问题2：Vault连接超时

现象：hvac.exceptions.VaultDown: Unable to connect to Vault server

排查步骤：

检查Vault服务状态：vault status
验证网络连通性：telnet vault 8200
更新客户端令牌：vault token renew

环境配置：

pip install fastapi==0.68.0 pydantic==1.8.2 python-multipart==0.0.5 hvac==0.11.2

代码验证方法：

import pytest
from fastapi.testclient import TestClient

client = TestClient(app)


def test_sensitive_masking():
    response = client.get("/users/1")
    assert "****" in response.json()["phone"]

以上内容通过实际案例演示了从字段标记到加密集成的完整数据脱敏流程，采用符合行业标准的安全实践，建议在生产环境中配合HTTPS和访问日志审计共同使用。

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长
，阅读完整的文章：如何在FastAPI中巧妙玩转数据脱敏，让敏感信息安全无忧？

往期文章归档：

免费好用的热门在线工具

登录查看全部

参与评论