FastAPI密码哈希与安全实践完全指南

📂 所属阶段：第四阶段 — 安全与认证（安全篇）
🔗 相关章节：FastAPI OAuth2与JWT鉴权 · FastAPI依赖注入系统

目录

密码安全基础

密码安全是Web应用程序的基石，错误的密码存储方式可能导致严重的安全漏洞。

核心原则

  1. 绝不存储明文密码
  2. 使用强哈希算法（bcrypt、Argon2）
  3. 每个用户使用唯一随机盐
  4. 定期更新哈希算法

错误示例

# ❌ 明文存储
def bad_register(email: str, password: str):
    query = f"INSERT INTO users (email, password) VALUES ('{email}', '{password}')"

# ❌ 简单哈希（MD5/SHA1）
import hashlib
def weak_hash(password: str) -> str:
    return hashlib.md5(password.encode()).hexdigest()

# ❌ 固定盐值
SALT = "myapp_salt"
def fixed_salt_hash(password: str) -> str:
    return hashlib.sha256((password + SALT).encode()).hexdigest()

正确示例

# ✅ 使用bcrypt安全哈希
from passlib.context import CryptContext

pwd_context = CryptContext(
    schemes=["bcrypt"], 
    deprecated="auto",
    bcrypt__rounds=12
)

def hash_password(password: str) -> str:
    return pwd_context.hash(password)

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

Passlib密码库详解

安装配置

pip install passlib[bcrypt] bcrypt

封装密码管理器

from passlib.context import CryptContext
from passlib.exc import UnknownHashError
import logging

class PasswordManager:
    def __init__(self):
        self.pwd_context = CryptContext(
            schemes=["bcrypt"],
            deprecated="auto",
            bcrypt__default_rounds=12
        )
        self.logger = logging.getLogger(__name__)
    
    def hash(self, password: str) -> str:
        try:
            return self.pwd_context.hash(password)
        except Exception as e:
            self.logger.error(f"哈希失败: {e}")
            raise
    
    def verify(self, plain: str, hashed: str) -> bool:
        try:
            return self.pwd_context.verify(plain, hashed)
        except UnknownHashError:
            return False
        except Exception as e:
            self.logger.error(f"验证失败: {e}")
            return False
    
    def needs_rehash(self, hashed: str) -> bool:
        return self.pwd_context.needs_update(hashed)

使用示例

manager = PasswordManager()

# 哈希密码
hashed = manager.hash("MySecurePassword123!")
print(f"哈希结果: {hashed}")

# 验证密码
print(manager.verify("MySecurePassword123!", hashed))  # True
print(manager.verify("WrongPassword", hashed))  # False

bcrypt算法深度解析

bcrypt基于Blowfish加密算法，具有内置盐值、可调节成本因子等特点。

成本因子调整

from passlib.context import CryptContext
import time

class CostOptimizer:
    def __init__(self, target_time: float = 0.1):
        self.target_time = target_time
    
    def find_optimal_cost(self) -> int:
        for cost in range(16, 4, -1):
            ctx = CryptContext(schemes=["bcrypt"], bcrypt__default_rounds=cost)
            start = time.time()
            ctx.hash("testpassword")
            duration = time.time() - start
            if duration <= self.target_time:
                return cost
        return 12

optimizer = CostOptimizer(target_time=0.2)
optimal_cost = optimizer.find_optimal_cost()
print(f"推荐成本因子: {optimal_cost}")

异步密码处理

import asyncio
from concurrent.futures import ThreadPoolExecutor

class AsyncPasswordManager(PasswordManager):
    def __init__(self, max_workers: int = 4):
        super().__init__()
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
    
    async def hash_async(self, password: str) -> str:
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(self.executor, self.hash, password)
    
    async def verify_async(self, plain: str, hashed: str) -> bool:
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(self.executor, self.verify, plain, hashed)

密码强度验证与策略

密码强度验证器

import re
from typing import List
from dataclasses import dataclass

@dataclass
class StrengthResult:
    is_strong: bool
    score: int
    feedback: List[str]

class PasswordValidator:
    def __init__(self):
        self.min_len = 8
        self.max_len = 128
        self.require_upper = True
        self.require_lower = True
        self.require_digit = True
        self.require_special = True
        self.special_chars = "!@#$%^&*(),.?\":{}|<>"
    
    def validate(self, password: str) -> StrengthResult:
        feedback = []
        score = 0
        
        # 长度检查
        if self.min_len <= len(password) <= self.max_len:
            score += 25
        else:
            feedback.append(f"密码长度应在{self.min_len}-{self.max_len}字符之间")
        
        # 大写字母
        if self.require_upper:
            if re.search(r'[A-Z]', password):
                score += 25
            else:
                feedback.append("需要至少一个大写字母")
        
        # 小写字母
        if self.require_lower:
            if re.search(r'[a-z]', password):
                score += 25
            else:
                feedback.append("需要至少一个小写字母")
        
        # 数字
        if self.require_digit:
            if re.search(r'\d', password):
                score += 25
            else:
                feedback.append("需要至少一个数字")
        
        # 特殊字符
        if self.require_special:
            if any(c in self.special_chars for c in password):
                score += 25
            else:
                feedback.append("需要至少一个特殊字符")
        
        # 防止满分超过100
        score = min(100, score)
        
        return StrengthResult(
            is_strong=score >= 80,
            score=score,
            feedback=feedback
        )

Pydantic验证

from pydantic import BaseModel, field_validator, EmailStr

class RegistrationRequest(BaseModel):
    email: EmailStr
    password: str
    confirm_password: str
    first_name: str
    last_name: str
    
    @field_validator("password")
    @classmethod
    def validate_password(cls, v: str) -> str:
        validator = PasswordValidator()
        result = validator.validate(v)
        if not result.is_strong:
            raise ValueError(f"密码强度不足: {', '.join(result.feedback)}")
        return v
    
    @field_validator("confirm_password")
    @classmethod
    def passwords_match(cls, v: str, info) -> str:
        if "password" in info.data and v != info.data["password"]:
            raise ValueError("两次密码不一致")
        return v

安全的用户注册流程

用户模型

from sqlalchemy import Column, Integer, String, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(255), unique=True, index=True, nullable=True)
    hashed_password = Column(String(255), nullable=False)
    first_name = Column(String(100), nullable=False)
    last_name = Column(String(100), nullable=False)
    is_active = Column(Boolean, default=True)
    is_verified = Column(Boolean, default=False)
    is_locked = Column(Boolean, default=False)
    failed_login_attempts = Column(Integer, default=0)
    created_at = Column(DateTime, default=func.now())

用户服务

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from typing import Optional

class UserService:
    def __init__(self, db: AsyncSession, password_manager: PasswordManager):
        self.db = db
        self.pwd_manager = password_manager
        self.lockout_threshold = 5
    
    async def register(
        self, email: str, password: str, first_name: str, last_name: str
    ) -> User:
        # 检查邮箱是否存在
        existing = await self.get_by_email(email)
        if existing:
            raise ValueError("邮箱已被注册")
        
        # 创建用户
        user = User(
            email=email,
            hashed_password=self.pwd_manager.hash(password),
            first_name=first_name,
            last_name=last_name
        )
        
        self.db.add(user)
        await self.db.commit()
        await self.db.refresh(user)
        
        return user
    
    async def login(self, email: str, password: str) -> Optional[User]:
        user = await self.get_by_email(email)
        if not user:
            return None
        
        if user.is_locked:
            raise ValueError("账户已锁定")
        
        if self.pwd_manager.verify(password, user.hashed_password):
            user.failed_login_attempts = 0
            await self.db.commit()
            return user
        else:
            user.failed_login_attempts += 1
            if user.failed_login_attempts >= self.lockout_threshold:
                user.is_locked = True
            await self.db.commit()
            return None
    
    async def get_by_email(self, email: str) -> Optional[User]:
        stmt = select(User).where(User.email == email)
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

密码安全最佳实践

安全配置清单

"""
✅ 技术层面
- 使用bcrypt/Argon2进行密码哈希
- 每个用户使用唯一随机盐
- 设置合理的成本因子
- 实施密码强度验证
- 定期更新哈希算法
- 账户锁定策略
- HTTPS传输
- 不记录密码到日志

✅ 业务层面
- 密码重置功能
- 邮箱验证机制
- 两因素认证
- 安全事件日志
- 定期安全审计
"""

密码泄露检测

import hashlib
import requests

class PasswordBreachDetector:
    def __init__(self):
        self.api_url = "https://api.pwnedpasswords.com/range/"
    
    def check(self, password: str) -> tuple[bool, int]:
        try:
            sha1 = hashlib.sha1(password.encode()).hexdigest().upper()
            prefix, suffix = sha1[:5], sha1[5:]
            
            response = requests.get(f"{self.api_url}{prefix}")
            response.raise_for_status()
            
            for line in response.text.splitlines():
                hash_suffix, count = line.split(':')
                if hash_suffix == suffix:
                    return True, int(count)
            
            return False, 0
        except Exception:
            return False, 0

安全中间件

from fastapi import Request, HTTPException, status
import time
from collections import defaultdict

class RateLimiter:
    def __init__(self):
        self.storage = defaultdict(list)
        self.window = 60
        self.max_requests = 100
    
    async def check(self, request: Request) -> bool:
        client_ip = request.client.host
        now = time.time()
        
        self.storage[client_ip] = [
            t for t in self.storage[client_ip]
            if now - t < self.window
        ]
        
        if len(self.storage[client_ip]) >= self.max_requests:
            return False
        
        self.storage[client_ip].append(now)
        return True

rate_limiter = RateLimiter()

总结

FastAPI中的密码安全实践提供了完整的安全认证解决方案：

  1. 密码哈希：使用bcrypt/Argon2等安全算法
  2. 强度验证：实施密码复杂度要求
  3. 安全存储：防彩虹表攻击的盐值机制
  4. 用户管理：完整的注册登录流程
  5. 安全防护：防暴力破解、速率限制

💡 关键要点：永远不要自己实现加密算法，使用经过安全审计的库。密码安全是Web应用的基石，必须给予足够重视。

🔗 扩展阅读