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