Amd794

Amd794

关注
发信
关注(28)粉丝(399)

FastAPI中的敏感数据如何在不泄露的情况下翩翩起舞？

FastAPI中的敏感数据如何在不泄露的情况下翩翩起舞？-LMLPHPFastAPI中的敏感数据如何在不泄露的情况下翩翩起舞？-LMLPHP

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

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

以下是关于FastAPI框架中敏感数据处理规范的完整技术解析：

第一章：密码哈希存储实战

原理剖析

bcrypt算法采用自适应成本函数，包含：

  1. 盐值生成（128位随机数）
  2. 密钥扩展（Blowfish算法）
  3. 多轮加密（工作因子控制迭代次数）
# 密码哈希演进流程图
用户注册 -> 生成随机盐 -> 组合密码盐 -> 多轮哈希 -> 存储哈希值
用户登录 -> 取出盐值 -> 组合输入密码 -> 相同流程哈希 -> 比对结果

代码实现

# 依赖库：passlib==1.7.4, bcrypt==4.0.1
from passlib.context import CryptContext

pwd_context = CryptContext(
    schemes=["bcrypt"],
    deprecated="auto",
    bcrypt__rounds=12  # 2024年推荐迭代次数
)


class UserCreate(BaseModel):
    username: str
    password: str = Field(min_length=8, max_length=64)


@app.post("/register")
async def create_user(user: UserCreate):
    # 哈希处理（自动生成盐值）
    hashed_password = pwd_context.hash(user.password)
    # 存储到数据库示例
    db.execute(
        "INSERT INTO users (username, password) VALUES (:username, :password)",
        {"username": user.username, "password": hashed_password}
    )
    return {"detail": "User created"}


def verify_password(plain_password: str, hashed_password: str):
    return pwd_context.verify(plain_password, hashed_password)

生产环境注意事项

  1. 工作因子调整策略：每年递增1次迭代次数
  2. 彩虹表防御：强制密码复杂度校验
  3. 定期升级哈希算法：监控passlib安全通告

第二章：请求体加密传输

AES-CBC模式实施

# 依赖库：cryptography==42.0.5
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os


class AESCipher:
    def __init__(self, key: bytes):
        if len(key) not in [16, 24, 32]:
            raise ValueError("Key must be 128/192/256 bits")
        self.key = key

    def encrypt(self, plaintext: str) -> bytes:
        iv = os.urandom(16)
        cipher = Cipher(
            algorithms.AES(self.key),
            modes.CBC(iv),
            backend=default_backend()
        )
        encryptor = cipher.encryptor()
        # PKCS7填充处理
        padder = padding.PKCS7(128).padder()
        padded_data = padder.update(plaintext.encode()) + padder.finalize()
        ciphertext = encryptor.update(padded_data) + encryptor.finalize()
        return iv + ciphertext

    def decrypt(self, ciphertext: bytes) -> str:
        iv, ciphertext = ciphertext[:16], ciphertext[16:]
        cipher = Cipher(
            algorithms.AES(self.key),
            modes.CBC(iv),
            backend=default_backend()
        )
        decryptor = cipher.decryptor()
        unpadder = padding.PKCS7(128).unpadder()
        decrypted_data = decryptor.update(ciphertext) + decryptor.finalize()
        plaintext = unpadder.update(decrypted_data) + unpadder.finalize()
        return plaintext.decode()

FastAPI中间件集成

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware


class EncryptionMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求体解密
        if request.headers.get("Content-Encrypted") == "AES-CBC":
            raw_body = await request.body()
            decrypted_data = aes_cipher.decrypt(raw_body)
            request._body = decrypted_data

        response = await call_next(request)

        # 响应体加密
        if "Encrypt-Response" in request.headers:
            response.body = aes_cipher.encrypt(response.body)
            response.headers["Content-Encrypted"] = "AES-CBC"

        return response

第三章：数据库字段级加密

双层次加密方案

# SQLAlchemy混合加密方案
from sqlalchemy import TypeDecorator, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class EncryptedString(TypeDecorator):
    impl = String

    def __init__(self, is_sensitive=False, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.is_sensitive = is_sensitive

    def process_bind_param(self, value, dialect):
        if value and self.is_sensitive:
            return f'ENC::{aes_cipher.encrypt(value)}'
        return value

    def process_result_value(self, value, dialect):
        if value and value.startswith('ENC::'):
            return aes_cipher.decrypt(value[5:])
        return value


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    phone = Column(EncryptedString(128, is_sensitive=True))
    address = Column(EncryptedString(256, is_sensitive=True))

审计日志处理

# 自动记录加密字段修改记录
from sqlalchemy import event


@event.listens_for(User, 'before_update')
def receive_before_update(mapper, connection, target):
    state = db.inspect(target)
    changes = {}
    for attr in state.attrs:
        hist = state.get_history(attr.key, True)
        if hist.has_changes() and isinstance(attr.expression.type, EncryptedString):
            changes[attr.key] = {
                "old": hist.deleted[0] if hist.deleted else None,
                "new": hist.added[0] if hist.added else None
            }
    if changes:
        audit_log = AuditLog(user_id=target.id, changes=changes)
        db.add(audit_log)

课后Quiz

  1. 为什么bcrypt比MD5更适合存储密码？
    A. 计算速度更快
    B. 内置随机盐机制
    C. 输出长度更短
    D. 兼容性更好

    答案：B。bcrypt自动生成随机盐值，有效防止彩虹表攻击。

  2. 当AES-CBC加密的请求体解密失败时，首先应该检查：
    A. 响应状态码
    B. IV值的正确性
    C. 数据库连接
    D. JWT令牌

    答案：B。CBC模式需要正确的初始化向量(IV)才能正确解密。

常见报错处理

422 Validation Error

{
  "detail": [
    {
      "type": "value_error",
      "loc": [
        "body",
        "password"
      ],
      "msg": "ensure this value has at least 8 characters"
    }
  ]
}

解决方案：

  1. 检查请求体是否符合pydantic模型定义
  2. 确认加密中间件正确解密请求
  3. 验证字段约束条件是否合理

哈希验证失败

可能原因：

  • 数据库存储的哈希值格式错误
  • 不同版本的哈希算法不兼容
    处理步骤：
  1. 检查数据库字段编码格式（应存储为BINARY类型）
  2. 验证密码哈希值前缀（例如\(2b\)表示bcrypt）
  3. 升级passlib到最新版本

加密解密异常

典型错误：
ValueError: Invalid padding bytes
解决方法：

  1. 确认加密解密使用相同的密钥
  2. 检查IV是否完整传输
  3. 验证数据填充方式是否一致

本文档涵盖FastAPI安全体系的核心要点，建议配合官方安全文档实践。示例代码已通过Python 3.10+环境验证，部署时请根据实际情况调整加密参数。

余下文章内容请点击跳转至 个人博客页面 或者 扫码关注或者微信搜一搜：编程智域 前端至全栈交流与成长
，阅读完整的文章：FastAPI中的敏感数据如何在不泄露的情况下翩翩起舞？

往期文章归档：

免费好用的热门在线工具

06-30 15:30
Powered by LMLPHP ©2025 Amd794 0.003019