From d5c498549b55ce9291f62ad02e37c0714cb9cf2b Mon Sep 17 00:00:00 2001
From: Your Name <your.email@example.com>
Date: Sat, 23 May 2026 16:22:36 +0800
Subject: [PATCH] docs: system development standard v1.0 - full coding rules
 for Python/FastAPI/Frontend

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 docs/project/system-development-standard.md | 499 ++++++++++++++++++++
 1 file changed, 499 insertions(+)
 create mode 100644 docs/project/system-development-standard.md

diff --git a/docs/project/system-development-standard.md b/docs/project/system-development-standard.md
new file mode 100644
index 00000000..c39cf609
--- /dev/null
+++ b/docs/project/system-development-standard.md
@@ -0,0 +1,499 @@
+# 系统开发完整标准 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 路由
+
+```python
+# ✅ 正确：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 数据库
+
+```python
+# ✅ 时间戳：必须带 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
+
+```python
+# ✅ 使用 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
+
+```python
+# ✅ 整段 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 凭据与加密
+
+```python
+# ✅ 存储：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 错误处理
+
+```python
+# ✅ 向上抛，让调用方决定
+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 规范
+
+```python
+# ✅ 所有 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 调用
+
+```javascript
+// ✅ 统一封装函数（含 Authorization + 错误处理）
+const data = await apiFetch("/api/resource/");
+
+// ❌ 禁止裸 fetch（漏 Authorization）
+const data = await fetch("/api/resource/");
+```
+
+### 4.2 XSS 防御（强制）
+
+```javascript
+// ✅ 文本内容用 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 静态资源
+
+```html
+<!-- ✅ 所有第三方 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 操作必须写）
+
+```python
+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 外部通知脱敏
+
+```python
+# 发 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 时间戳规范
+
+```python
+# ✅ 所有 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 外键删除策略
+
+```python
+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 命名规范
+
+```python
+# 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）
+
+---
+
+## 八、数据库迁移规范
+
+```python
+# ✅ 幂等迁移（每次启动安全重跑）
+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 防重复
+
+```python
+# ✅ 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 后台任务结构
+
+```python
+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 防区时
+
+```python
+# ✅ 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 必含字段
+
+```markdown
+# YYYY-MM-DD 主题
+
+**日期**: YYYY-MM-DD
+**动机**: 为什么要做这个变更
+**变更摘要**: 做了什么
+**涉及文件**: 列出主要文件
+**是否需迁移**: 是/否（若是，说明步骤）
+**验证方式**: 如何确认变更生效
+```
+
+---
+
+## 十一、禁止代码示例
+
+```python
+# ❌ 技术债（禁止提交）
+# 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，用于约束代码生成、代码审查、架构评审任务。*