Nexus/standards/system-development-standard.md

系统开发完整标准 v1.0

版本: 1.0 · 2026-05-23
适用: Python/FastAPI 后端 + 前端 全栈项目
来源: perfect-implementation.mdc + nexus-*规则 + Phase 1–4 审计实践提炼
性质: 强制规范，不得降级或妥协

---

零、实现铁律

❌ 不允许为了实现而实现   — 每行代码必须有明确目的
❌ 不允许降级实现         — 不以「先用着」为由降低安全/架构/质量
❌ 不允许妥协实现         — 发现问题必须根治，不得绕过
❌ 不允许留技术债         — TODO / FIXME / 临时方案禁止遗留
✅ 无法完美实现时必须停   — 告知用户确认方案后再换路，不擅自从权

---

一、开发执行顺序（每次必走）

需求确认 → 设计文档 → 技术文档 → 实现 → 测试验证 → changelog → 更新项目记忆文档

阶段	产出路径	必含内容
设计文档	docs/design/specs/YYYY-MM-DD-<topic>-design.md	背景与目标、方案对比、选定方案及理由、接口/数据模型、安全约束、验收标准
技术文档	docs/design/plans/YYYY-MM-DD-<topic>.md	涉及文件清单、实现步骤、依赖/配置变更、测试要点、回滚方式
Changelog	docs/changelog/YYYY-MM-DD-<主题>.md	日期、变更摘要、动机、涉及文件、是否需迁移/重启、验证方式

新功能/架构变更/非平凡重构：先设计文档，再写代码。
改动时已有模块缺文档：一并补全，不得只改代码不留档。
禁止：把 changelog 只写进 commit message。

---

二、架构分层（禁止跨层直连）

HTTP 路由层  →  Service 层  →  Repository 层  →  Domain 模型
   (api/)       (services/)     (*_repo.py)        (models/)

基础设施层（database / redis / ssh / telegram）仅被 Service 层或 Repository 层调用
前端（web/app/）仅通过 HTTP API 与后端通信，不直接操作 DB/Redis

层	职责	禁止
API 路由层	请求校验、鉴权 Depends、调用 Service、返回响应	写 SQL、直接操作 Redis
Service 层	业务编排、事务边界、审计写入	直接 import FastAPI 依赖
Repository 层	DB 读写、分页、批量操作	业务逻辑
Domain 模型	ORM 字段定义、关系	任何业务逻辑、FastAPI 依赖
基础设施层	DB 连接池、Redis 客户端、SSH 池、外部 HTTP	业务逻辑

---

三、后端编码规范（Python / FastAPI）

3.1 API 路由

# ✅ 正确：Pydantic 校验 + JWT 鉴权 + 结构化错误
@router.post("/resource", response_model=dict, status_code=201)
async def create_resource(
    payload: ResourceCreate,                    # Pydantic 自动校验
    admin: Admin = Depends(get_current_admin),  # JWT 鉴权
    service: ResourceService = Depends(...),
    db: AsyncSession = Depends(get_db),
):
    ...
    raise HTTPException(status_code=422, detail="具体原因")  # 结构化错误

# ❌ 禁止：裸 except、无鉴权、无 Pydantic
@router.post("/resource")
async def create(data: dict):   # 无 Pydantic，无鉴权
    try:
        ...
    except:
        pass  # 禁止静默吞错

3.2 鉴权矩阵（强制）

调用方	机制	备注
管理端 API	Depends(get_current_admin) JWT Bearer	所有业务路由必须
Agent	X-API-Key header + secrets.compare_digest	防时序攻击
WebSocket	JWT query param ?token=	ADR-011
安装向导	无 JWT	仅安装模式且安装后 403

3.3 数据库

# ✅ 时间戳：必须带 timezone
datetime.now(timezone.utc)         # ✅
datetime.now()                     # ❌ naive，与 aware 比较 → TypeError

# ✅ 分页在 Repository 层
async def get_paginated(offset, limit) -> tuple[list, int]:
    count = await session.execute(select(func.count(Model.id)))
    data = await session.execute(select(Model).offset(offset).limit(limit))

# ✅ 字段长度：加密后的值须 Text 而非 String(N)
ssh_key_private = Column(Text)     # RSA 4096 Fernet 后 ~4300 chars
encrypted_pw    = Column(String(500))  # 短密码 Fernet 后 ~100 chars，可 String

# ✅ 外键删除策略：明确声明
ForeignKey("servers.id", ondelete="CASCADE")   # 主记录删 → 关联删
ForeignKey("scripts.id", ondelete="SET NULL")  # 主记录删 → 关联置 null

3.4 Redis

# ✅ 使用 pipeline 保证 INCR + EXPIRE 原子性
pipe = redis.pipeline()
pipe.incr(key)
pipe.expire(key, window_seconds)
count, _ = await pipe.execute()

# ❌ 非原子（崩溃在中间 → key 无 TTL 永久存在）
count = await redis.incr(key)
await redis.expire(key, window_seconds)  # 若崩溃 → 永久存在

3.5 SSH / Shell

# ✅ 整段 key=value 必须整体 quote
cmd = f"echo {shlex.quote(f'{key}={value}')}"   # ✅
cmd = f"echo {shlex.quote(key)}={shlex.quote(value)}"  # ❌ 等号在引号外

# ✅ shlex.quote 结果不放进双引号（引号嵌套变字面量）
f"{shlex.quote(path)}.pid"     # ✅ 产生 '/path/to/file'.pid
f'"{shlex.quote(path)}.pid"'   # ❌ 产生 "'/path/to/file'.pid"（单引号字面量）

# ✅ 多行注入防御：剥离换行
safe_value = str(value).replace('\n',' ').replace('\r',' ').replace('\x00','')

3.6 凭据与加密

# ✅ 存储：Fernet 加密
password_col = encrypt_value(plaintext)

# ✅ 响应：只返回布尔标记
{"password_set": bool(server.password)}   # ✅
{"password": server.password}              # ❌ 禁止返回密文或明文

# ✅ 变量替换：两轮替换防嵌套
SENTINELS = {"$DB_PASS": "\x00PASS\x00", "$DB_USER": "\x00USER\x00"}
VALUES = {"\x00PASS\x00": real_password, "\x00USER\x00": real_user}
for placeholder, sentinel in SENTINELS.items():
    cmd = cmd.replace(placeholder, sentinel)
for sentinel, value in VALUES.items():
    cmd = cmd.replace(sentinel, value)

# ✅ 日志：脱敏后记录
logger.info(f"Redis: {_mask_url(settings.REDIS_URL)}")  # 非 settings.REDIS_URL 原文

3.7 错误处理

# ✅ 向上抛，让调用方决定
def decrypt_value(ciphertext):
    try:
        return fernet.decrypt(...)
    except Exception as e:
        raise ValueError("解密失败") from e  # ✅ 不返回密文，不静默

# ❌ 静默吞错（禁止）
    except Exception:
        return ciphertext   # ❌ 返回密文 = 信息泄露
        return None         # ❌ 调用方看不到错误

# ✅ gather 异常必须检查
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        logger.error("Task failed: %s", r)

3.8 Datetime 规范

# ✅ 所有 datetime 带 timezone
from datetime import datetime, timezone
now = datetime.now(timezone.utc)              # ✅

# ✅ MySQL 返回 naive，比较前统一
expires = record.expires_at                   # naive（MySQL 存储无时区）
now_cmp = datetime.now(timezone.utc)
if expires.tzinfo is None:
    now_cmp = now_cmp.replace(tzinfo=None)    # 统一为 naive 再比较
if expires < now_cmp: ...

# ✅ Cron 解析防除零
step = int(part[2:])
if step == 0:
    return False   # 无效 step，跳过

---

四、前端编码规范

4.1 API 调用

// ✅ 统一封装函数（含 Authorization + 错误处理）
const data = await apiFetch("/api/resource/");

// ❌ 禁止裸 fetch（漏 Authorization）
const data = await fetch("/api/resource/");

4.2 XSS 防御（强制）

// ✅ 文本内容用 textContent 或 Alpine 绑定
el.textContent = serverName;

// ✅ 必须插入 HTML 时用 esc()
el.innerHTML = `<b>${esc(serverName)}</b>`;

// ❌ 禁止
el.innerHTML = `<b>${serverName}</b>`;             // 未 esc
el.innerHTML = `<button onclick="fn('${val}')">`;  // onclick 注入

// ✅ data 属性安全传值
`<button data-id="${esc(id)}" @click="handle($el.dataset.id)">`

4.3 静态资源

<!-- ✅ 所有第三方 JS/CSS 自托管（/app/vendor/）或带 SRI -->
<script src="/app/vendor/alpinejs.min.js"></script>

<!-- ❌ 无 SRI 的 CDN -->
<script src="https://cdn.jsdelivr.net/npm/alpinejs@3"></script>

4.4 会话管理

• JWT 存 localStorage，到期前 2 分钟自动 refresh
• 401 响应：retry 一次，失败则 _logout() 跳登录页
• 8 小时无活动强制退出（与后端 updated_at 校验对齐）

---

五、安全规范

5.1 密钥管理

密钥	规则
SECRET_KEY / API_KEY / ENCRYPTION_KEY	只在 .env，不进 git，不可从 DB settings 覆盖，UI 不展示
Agent api_key	secrets.token_urlsafe(32)；compare_digest 校验；启动时非空强制退出
SSH 私钥	Fernet 加密后存 Text 列；响应只返回 ssh_key_private_set: bool
JWT	HS256，含 exp/sub/updated claims；密码改后 token_version 递增使旧 token 失效

5.2 审计日志（所有 CUD 操作必须写）

await audit_repo.create(AuditLog(
    admin_username=admin.username,
    action="create_server",
    target_type="server",
    target_id=created.id,
    detail=f"name={created.name}",
    ip_address=request.client.host if request.client else "",
))

5.3 外部通知脱敏

# 发 Telegram / Slack / 邮件前必须脱敏
safe_msg = sanitize_external_message(str(exception))  # 去除 URL / password / key
await send_telegram_system_alert(safe_msg)
# 原始详情只写本地日志
logger.error("Full error: %s", exception, exc_info=True)

5.4 登录防暴破

• 失败计数（同 username + IP）→ 15 分钟锁定 → HTTP 429
• Refresh token 重用检测：旧 token 作废 + token_version 递增 + 审计日志

5.5 危险命令策略

• 中心管控面（server/api/）：阻断危险命令，返回 400
• 子机 Agent（web/agent/）：记录日志，架构上无法拦截（terminal 字符流）

---

六、数据模型规范

6.1 列类型选择

数据	类型	理由
短文本（<255 chars）	String(N)	索引高效
加密后的短密码	String(500)	Fernet 加密后 ~100-300 chars
SSH 私钥（加密后）	Text	RSA 4096 Fernet 后 ~4300 chars
JSON 数组/对象	JSON 或 Text（存 json.dumps）	视 DB 版本
命令输出 / 详情	Text	无长度限制

6.2 时间戳规范

# ✅ 所有 DateTime 列用 aware callable（不用 onupdate=func.now()）
created_at = Column(DateTime, default=lambda: datetime.now(timezone.utc))
updated_at = Column(DateTime, default=lambda: datetime.now(timezone.utc),
                    onupdate=lambda: datetime.now(timezone.utc))

6.3 外键删除策略

ondelete="CASCADE"   # 主记录删 → 关联记录也删（日志、会话）
ondelete="SET NULL"  # 主记录删 → 关联置 null（软依赖）

6.4 索引规范

• 高频过滤字段必须加索引：WHERE status = X AND created_at >= Y
• 复合索引列顺序：高选择性列在前
• 迁移幂等：CREATE INDEX IF NOT EXISTS 或 try/except duplicate

---

七、测试规范

7.1 覆盖要求

新 endpoint	测试路径
认证失败	无 token → 401
参数校验失败	必填字段缺失 → 422
成功路径	正常入参 → 正确响应
边界情况	空列表 / 极值

7.2 命名规范

# tests/test_<模块名>.py
def test_<函数名>_<场景>_<预期结果>():
    ...

# 示例
def test_create_server_missing_domain_returns_422(): ...
def test_decrypt_value_fernet_fails_raises_valueerror(): ...

7.3 禁止

• 测试中硬编码生产密钥
• 跳过鉴权测试（skip 除非有充分理由）
• 测试依赖真实数据库（使用 mock 或 test fixture）

---

八、数据库迁移规范

# ✅ 幂等迁移（每次启动安全重跑）
try:
    await conn.execute(text("ALTER TABLE t ADD COLUMN c VARCHAR(100)"))
except Exception as e:
    if "duplicate" in str(e).lower() or "already exists" in str(e).lower():
        pass  # 列已存在，跳过
    else:
        raise  # 真正的错误必须抛出

# ✅ 索引幂等
try:
    await conn.execute(text("CREATE INDEX idx_name ON table (col)"))
except Exception as e:
    if "duplicate" in str(e).lower():
        pass
    else:
        raise

---

九、后台任务规范

9.1 多 Worker 防重复

# ✅ Redis 分布式锁：只有 primary worker 运行后台任务
acquired = await redis.set(LOCK_KEY, worker_id, nx=True, ex=TTL)
if acquired:
    task = asyncio.create_task(my_loop())

# ✅ 锁丢失时取消任务
if lock_lost:
    for task in _background_tasks:
        task.cancel()

9.2 后台任务结构

async def my_background_loop():
    logger.info("Loop started (interval=%ss)", INTERVAL)
    while True:
        await asyncio.sleep(INTERVAL)
        try:
            async with AsyncSessionLocal() as session:
                await do_work(session)
        except asyncio.CancelledError:
            break
        except Exception as e:
            logger.error("Loop error: %s", e)
            # 继续下一个周期，不终止整个 loop

9.3 Datetime 防区时

# ✅ MySQL 返回 naive，统一处理
last_run = record.last_run_at              # naive
now_naive = datetime.now(timezone.utc).replace(tzinfo=None)
if last_run.tzinfo is not None:
    last_run = last_run.replace(tzinfo=None)
elapsed = (now_naive - last_run).total_seconds()

---

十、文档交付标准

每次 PR / 功能完成必须包含

[ ] docs/changelog/YYYY-MM-DD-<主题>.md 已创建
[ ] 涉及新 API：schemas.py 有 Pydantic 模型
[ ] 涉及新 DB 表/列：domain/models/__init__.py 已更新 + 迁移脚本幂等
[ ] 涉及安全变更：audit-log 写入已验证
[ ] 项目记忆文档（CLAUDE.md / AGENTS.md）已更新实施状态

Changelog 必含字段

# YYYY-MM-DD 主题

**日期**: YYYY-MM-DD
**动机**: 为什么要做这个变更
**变更摘要**: 做了什么
**涉及文件**: 列出主要文件
**是否需迁移**: 是/否（若是，说明步骤）
**验证方式**: 如何确认变更生效

---

十一、禁止代码示例

# ❌ 技术债（禁止提交）
# TODO: fix later
result = "workaround"

# ❌ 静默失败
except Exception:
    return None

# ❌ 无文档大改
# （新增整模块却无 docs/design 或 docs/changelog 记录）

# ❌ tz-aware/naive 混用
datetime.now() - record.created_at  # 若一个 aware 一个 naive → TypeError

# ❌ SQL 注入
await session.execute(text(f"SELECT * FROM t WHERE id = {user_input}"))

# ❌ 命令注入
cmd = f"rsync {source} {target}"  # source/target 未 shlex.quote

# ❌ 密钥返回 API
{"api_key": settings.API_KEY}     # 禁止返回完整密钥

# ❌ 日志泄密
logger.info(f"DB: {settings.DATABASE_URL}")  # URL 含密码

---

本规范可直接发给 AI，用于约束代码生成、代码审查、架构评审任务。