Codex Release Bot
500 lines
15 KiB
Markdown
500 lines
15 KiB
Markdown
# 系统开发完整标准 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，用于约束代码生成、代码审查、架构评审任务。*
|