TJWaterServerBinary/DEPLOYMENT.md

部署和集成指南

本文档说明如何将新的安全功能集成到现有系统中。

📦 已完成的功能

1. 数据加密模块

• ✅ app/core/encryption.py - Fernet 对称加密实现
• ✅ 支持敏感数据加密/解密
• ✅ 密钥管理和生成工具

2. 用户认证系统

• ✅ app/domain/models/role.py - 用户角色枚举 (ADMIN/OPERATOR/USER/VIEWER)
• ✅ app/domain/schemas/user.py - 用户数据模型和验证
• ✅ app/infra/repositories/user_repository.py - 用户数据访问层
• ✅ app/api/v1/endpoints/auth.py - 注册/登录/刷新Token接口
• ✅ app/auth/dependencies.py - 认证依赖项
• ✅ migrations/001_create_users_table.sql - 用户表迁移脚本

3. 权限控制系统

• ✅ app/auth/permissions.py - RBAC 权限控制装饰器
• ✅ app/api/v1/endpoints/user_management.py - 用户管理接口示例
• ✅ 支持基于角色的访问控制
• ✅ 支持资源所有者检查

4. 审计日志系统

• ✅ app/core/audit.py - 审计日志核心功能
• ✅ app/domain/schemas/audit.py - 审计日志数据模型
• ✅ app/infra/repositories/audit_repository.py - 审计日志数据访问层
• ✅ app/api/v1/endpoints/audit.py - 审计日志查询接口
• ✅ app/infra/audit/middleware.py - 自动审计中间件
• ✅ migrations/002_create_audit_logs_table.sql - 审计日志表迁移脚本

5. 文档和测试

• ✅ SECURITY_README.md - 完整的使用文档
• ✅ .env.example - 环境变量配置模板
• ✅ tests/test_encryption.py - 加密功能测试

---

🔧 集成步骤

步骤 1: 环境配置

1. 复制环境变量模板：

cp .env.example .env

2. 生成密钥并填写 .env：

# JWT 密钥
openssl rand -hex 32

# 加密密钥
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

3. 编辑 .env 填写所有必需的配置项。

步骤 2: 数据库迁移

执行数据库迁移脚本：

# 方法 1: 使用 psql 命令
psql -U postgres -d tjwater -f migrations/001_create_users_table.sql
psql -U postgres -d tjwater -f migrations/002_create_audit_logs_table.sql

# 方法 2: 在 psql 交互界面
psql -U postgres -d tjwater
\i migrations/001_create_users_table.sql
\i migrations/002_create_audit_logs_table.sql

验证表已创建：

-- 检查用户表
SELECT * FROM users;

-- 检查审计日志表
SELECT * FROM audit_logs;

步骤 3: 更新 main.py

在 app/main.py 中集成新功能：

from fastapi import FastAPI
from app.core.config import settings
from app.infra.audit.middleware import AuditMiddleware

app = FastAPI(title=settings.PROJECT_NAME)

# 1. 添加审计中间件（可选）
app.add_middleware(AuditMiddleware)

# 2. 注册路由
from app.api.v1.endpoints import auth, user_management, audit

app.include_router(
    auth.router,
    prefix=f"{settings.API_V1_STR}/auth",
    tags=["认证"]
)

app.include_router(
    user_management.router,
    prefix=f"{settings.API_V1_STR}/users",
    tags=["用户管理"]
)

app.include_router(
    audit.router,
    prefix=f"{settings.API_V1_STR}/audit",
    tags=["审计日志"]
)

# 3. 确保数据库在启动时初始化
@app.on_event("startup")
async def startup_event():
    # 初始化数据库连接池
    from app.infra.db.postgresql.database import Database
    global db
    db = Database()
    db.init_pool()
    await db.open()

@app.on_event("shutdown")
async def shutdown_event():
    # 关闭数据库连接
    await db.close()

步骤 4: 保护现有接口

方法 1: 为路由添加全局依赖

from app.auth.dependencies import get_current_active_user

# 为整个路由器添加认证
router = APIRouter(dependencies=[Depends(get_current_active_user)])

方法 2: 为单个端点添加依赖

from app.auth.permissions import require_role, get_current_admin
from app.domain.models.role import UserRole

@router.get("/data")
async def get_data(
    current_user = Depends(require_role(UserRole.USER))
):
    """需要 USER 及以上角色"""
    return {"data": "protected"}

@router.delete("/data/{id}")
async def delete_data(
    id: int,
    current_user = Depends(get_current_admin)
):
    """仅管理员可访问"""
    return {"message": "deleted"}

步骤 5: 添加审计日志

自动审计（推荐）

使用中间件自动记录（已在 main.py 中添加）：

app.add_middleware(AuditMiddleware)

手动审计

在关键业务逻辑中手动记录：

from app.core.audit import log_audit_event, AuditAction

@router.post("/important-action")
async def important_action(
    data: dict,
    request: Request,
    current_user = Depends(get_current_active_user)
):
    # 执行业务逻辑
    result = do_something(data)
    
    # 记录审计日志
    await log_audit_event(
        action=AuditAction.UPDATE,
        user_id=current_user.id,
        username=current_user.username,
        resource_type="important_resource",
        resource_id=str(result.id),
        ip_address=request.client.host,
        request_data=data
    )
    
    return result

步骤 6: 更新 auth/dependencies.py

确保 get_db() 函数正确获取数据库实例：

async def get_db() -> Database:
    """获取数据库实例"""
    # 方法 1: 从 main.py 导入
    from app.main import db
    return db
    
    # 方法 2: 从 FastAPI app.state 获取
    # from fastapi import Request
    # def get_db_from_request(request: Request):
    #     return request.app.state.db

---

🧪 测试

1. 测试加密功能

python tests/test_encryption.py

2. 测试 API

启动服务器：

uvicorn app.main:app --reload

访问交互式文档：

• Swagger UI: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

3. 测试登录

curl -X POST "http://localhost:8000/api/v1/auth/login" \
  -H "Content-Type: application/x-www-form-urlencoded" \
  -d "username=admin&password=admin123"

4. 测试受保护接口

TOKEN="your-access-token"
curl -X GET "http://localhost:8000/api/v1/auth/me" \
  -H "Authorization: Bearer $TOKEN"

---

🔄 迁移现有接口

原有硬编码认证

旧代码 (app/api/v1/endpoints/auth.py):

AUTH_TOKEN = "567e33c876a2"

async def verify_token(authorization: str = Header()):
    token = authorization.split(" ")[1]
    if token != AUTH_TOKEN:
        raise HTTPException(status_code=403)

新代码 (已更新):

from app.auth.dependencies import get_current_active_user

@router.get("/protected")
async def protected_route(
    current_user = Depends(get_current_active_user)
):
    return {"user": current_user.username}

更新其他端点

搜索项目中使用旧认证的地方：

grep -r "AUTH_TOKEN" app/
grep -r "verify_token" app/

替换为新的依赖注入系统。

---

📋 检查清单

部署前检查：

• 环境变量已配置（.env）
• 数据库迁移已执行
• 默认管理员账号可登录
• JWT Token 可正常生成和验证
• 权限控制正常工作
• 审计日志正常记录
• 加密功能测试通过
• API 文档可访问

---

⚠️ 注意事项

1. 向后兼容性

保留了简化版登录接口 /auth/login/simple 以兼容旧客户端：

@router.post("/login/simple")
async def login_simple(username: str, password: str):
    # 验证并返回 Token
    ...

2. 数据库连接

确保在 app/auth/dependencies.py 中 get_db() 函数能正确获取数据库实例。

3. 密钥安全

• ❌ 不要提交 .env 文件到版本控制
• ✅ 在生产环境使用环境变量或密钥管理服务
• ✅ 定期轮换 JWT 密钥

4. 性能考虑

• 审计中间件会增加每个请求的处理时间（约 5-10ms）
• 对高频接口可考虑异步记录审计日志
• 定期清理或归档旧的审计日志

---

🐛 故障排查

问题 1: 导入错误

ImportError: cannot import name 'db' from 'app.main'

解决: 确保在 app/main.py 中定义了全局 db 对象。

问题 2: 认证失败

401 Unauthorized: Could not validate credentials

检查:

1. Token 是否正确设置在 Authorization: Bearer {token} header
2. Token 是否过期
3. SECRET_KEY 是否配置正确

问题 3: 数据库连接失败

psycopg.OperationalError: connection failed

检查:

1. PostgreSQL 是否运行
2. .env 中数据库配置是否正确
3. 数据库是否存在

---

📞 技术支持

详细文档请参考：

• SECURITY_README.md - 安全功能使用指南
• migrations/ - 数据库迁移脚本
• app/domain/schemas/ - 数据模型定义