From b866e944ab18c0aaeb6e3d572950b69cf04493e0 Mon Sep 17 00:00:00 2001 From: Nexus Deploy Date: Thu, 4 Jun 2026 15:57:49 +0800 Subject: [PATCH] fix: Code Review P0/P1 batches 1-3 and single-operator risk acceptance Apply sync/install/auth/schedule/retry/agent/settings fixes from full code review; document accepted WS and Agent legacy risks for solo ops. Co-authored-by: Cursor --- .../2026-06-04-code-review-p0-p1-fixes.md | 48 +++++++ .../2026-06-04-code-review-p1-p2-batch2.md | 31 ++++ .../2026-06-04-code-review-p1-p2-batch3.md | 41 ++++++ ...026-06-04-code-review-production-deploy.md | 60 ++++++++ .../2026-06-04-code-review-p0-p1-fixes.md | 43 ++++++ .../2026-06-04-code-review-p1-p2-batch2.md | 43 ++++++ .../2026-06-04-code-review-p1-p2-batch3.md | 50 +++++++ ...6-06-04-risk-acceptance-single-operator.md | 32 +++++ docs/project/AI-HANDOFF-2026-05-23.md | 21 ++- docs/project/AI-HANDOFF-2026-06-03.md | 21 +++ .../risk-acceptance-single-operator.md | 126 ++++++++++++++++ .../src/composables/push/usePushProgress.ts | 6 + frontend/src/composables/useServerList.ts | 6 +- frontend/src/pages/CommandsPage.vue | 6 +- frontend/src/pages/CredentialsPage.vue | 16 ++- frontend/src/pages/DashboardPage.vue | 7 +- frontend/src/pages/LoginPage.vue | 3 +- frontend/src/pages/ScriptsPage.vue | 6 +- frontend/src/pages/SettingsPage.vue | 87 +++++++++-- frontend/src/types/api.ts | 1 + server/api/agent.py | 35 +++-- server/api/install.py | 24 ++++ server/api/settings.py | 135 +++++++++--------- server/api/sync_v2.py | 38 ++--- server/api/websocket.py | 4 +- server/api/webssh.py | 18 ++- server/application/services/auth_service.py | 54 ++++++- server/application/services/script_service.py | 26 +++- server/application/services/sync_engine_v2.py | 100 +++++++------ server/background/retry_runner.py | 61 +++++--- server/background/schedule_runner.py | 108 ++++++++------ server/main.py | 24 +++- server/utils/posix_paths.py | 27 ++++ tests/test_retry_runner_outcome.py | 12 ++ tests/test_security_unit.py | 30 ++++ web/app/install.html | 3 + 36 files changed, 1111 insertions(+), 242 deletions(-) create mode 100644 docs/audit/2026-06-04-code-review-p0-p1-fixes.md create mode 100644 docs/audit/2026-06-04-code-review-p1-p2-batch2.md create mode 100644 docs/audit/2026-06-04-code-review-p1-p2-batch3.md create mode 100644 docs/audit/2026-06-04-code-review-production-deploy.md create mode 100644 docs/changelog/2026-06-04-code-review-p0-p1-fixes.md create mode 100644 docs/changelog/2026-06-04-code-review-p1-p2-batch2.md create mode 100644 docs/changelog/2026-06-04-code-review-p1-p2-batch3.md create mode 100644 docs/changelog/2026-06-04-risk-acceptance-single-operator.md create mode 100644 docs/project/risk-acceptance-single-operator.md diff --git a/docs/audit/2026-06-04-code-review-p0-p1-fixes.md b/docs/audit/2026-06-04-code-review-p0-p1-fixes.md new file mode 100644 index 00000000..1b579438 --- /dev/null +++ b/docs/audit/2026-06-04-code-review-p0-p1-fixes.md @@ -0,0 +1,48 @@ +# 审计 — 2026-06-04 Code Review P0/P1 修复 + +## Step 3 规则扫描（H） + +| H | 规则 | 文件 | 结论 | +|---|------|------|------| +| H1 | PY-08 静默吞错 | sync_engine_v2, auth_service | SAFE — 保留 logger，无空 pass | +| H2 | SEC 安装无 JWT | install.py | FIXED — install_token + compare_digest | +| H3 | SEC 路径遍历 | posix_paths.py | FIXED — 统一 resolve_nexus_push_source_path | +| H4 | CONC AsyncSession | sync_engine_v2.py | FIXED — _db_lock 串行 DB | +| H5 | CONC refresh 双花 | auth_service.py | FIXED — Redis Lua SREM+SADD | +| H6 | FE 开放重定向 | LoginPage.vue | FIXED | +| H7 | FE WS 泄漏 | usePushProgress.ts | FIXED | + +## Closure + +| H | 判定 | 依据 | +|---|------|------| +| H1 | SAFE | except 均 logging | +| H2 | FIXED | install.py:572+ `_verify_install_token` | +| H3 | FIXED | posix_paths.py:103-114 | +| H4 | FIXED | sync_engine_v2.py `_db_lock` | +| H5 | FIXED | auth_service `_rotate_refresh_token` Lua | +| H6 | FIXED | LoginPage redirect 校验 | +| H7 | FIXED | disconnectProgressWs + 4001 logout | + +## 改动文件清单 + +- server/application/services/sync_engine_v2.py +- server/application/services/auth_service.py +- server/api/install.py +- server/api/sync_v2.py +- server/utils/posix_paths.py +- server/main.py +- server/background/schedule_runner.py +- web/app/install.html +- frontend/src/pages/LoginPage.vue +- frontend/src/composables/push/usePushProgress.ts +- tests/test_security_unit.py + +## DoD + +- [x] ruff 零错误（改动文件） +- [x] import server.main 成功 +- [x] pytest test_security_unit 通过 +- [x] frontend type-check 通过 +- [x] changelog ≥10 行 +- [ ] 生产部署与健康检查（待用户推送后执行） diff --git a/docs/audit/2026-06-04-code-review-p1-p2-batch2.md b/docs/audit/2026-06-04-code-review-p1-p2-batch2.md new file mode 100644 index 00000000..1fea28f5 --- /dev/null +++ b/docs/audit/2026-06-04-code-review-p1-p2-batch2.md @@ -0,0 +1,31 @@ +# 审计 — 2026-06-04 Code Review P1/P2 批次 2 + +## Step 3 规则扫描 + +| H | 规则 | 结论 | +|---|------|------| +| H1 | WebSSH tv | FIXED auth_service + webssh | +| H2 | retry 重复执行 | FIXED SKIP LOCKED | +| H3 | schedule 重复触发 | FIXED FOR UPDATE | +| H4 | Agent fail-open | FIXED agent.py | +| H5 | Bing DoS | FIXED cache-only GET + admin POST sync | +| H6 | install status 侦察 | FIXED 404 when locked | +| H7 | FE 静默失败 | FIXED 5 处 snackbar | + +## Closure + +| H | 判定 | 依据 | +|---|------|------| +| H1-H7 | FIXED | 见 changelog 与对应文件 diff | + +## 改动文件清单 + +见 `docs/changelog/2026-06-04-code-review-p1-p2-batch2.md` + +## DoD + +- [x] ruff 零错误 +- [x] import server.main +- [x] pytest 28 passed +- [x] vue-tsc 通过 +- [x] changelog ≥10 行 diff --git a/docs/audit/2026-06-04-code-review-p1-p2-batch3.md b/docs/audit/2026-06-04-code-review-p1-p2-batch3.md new file mode 100644 index 00000000..08a4efe7 --- /dev/null +++ b/docs/audit/2026-06-04-code-review-p1-p2-batch3.md @@ -0,0 +1,41 @@ +# 审计 — 2026-06-04 Code Review P1/P2 批次 3 + +## Step 3 规则扫描 + +| H | 规则 | 结论 | +|---|------|------| +| H1 | 脚本并行 AsyncSession | FIXED script_service 预加载 server_map | +| H2 | Settings 敏感项泄露 | FIXED value="" + value_set | +| H3 | 空 PUT 清空 token | FIXED 敏感项空值 no-op | +| H4 | Agent unknown server + global | FIXED 必须存在 server；无 key 才 legacy global | +| H5 | WS 关闭码混用 | FIXED 无效 JWT 4401 | +| H6 | FE Telegram token 明文进表单 | FIXED draft + reveal | + +## Closure + +| H | 判定 | 依据 | +|---|------|------| +| H1 | FIXED | `_dispatch_to_servers` / `_attach_remote_logs` 并行前 `server_map` | +| H2 | FIXED | `_format_setting_response` SENSITIVE_KEYS | +| H3 | FIXED | `set_setting` 337-342 行空值分支 | +| H4 | FIXED | `_verify_server_api_key` server 不存在 return False | +| H5 | FIXED | websocket.py / webssh.py 4401 | +| H6 | FIXED | SettingsPage.vue telegramTokenDraft + reveal-token | + +## 改动文件清单 + +- `server/application/services/script_service.py` +- `server/api/settings.py` +- `server/api/agent.py` +- `server/api/websocket.py` +- `server/api/webssh.py` +- `frontend/src/pages/SettingsPage.vue` +- `frontend/src/types/api.ts` + +## DoD + +- [x] ruff 零错误（批次 3 文件） +- [x] import server.main +- [x] pytest 28 passed +- [x] vue-tsc 通过 +- [x] changelog ≥10 行 diff --git a/docs/audit/2026-06-04-code-review-production-deploy.md b/docs/audit/2026-06-04-code-review-production-deploy.md new file mode 100644 index 00000000..0b2d4c08 --- /dev/null +++ b/docs/audit/2026-06-04-code-review-production-deploy.md @@ -0,0 +1,60 @@ +# 审计 — 2026-06-04 Code Review 生产部署 + +## Step 3 规则扫描 + +| H | 规则 | 结论 | +|---|------|------| +| H1 | P0/P1 batch1 | FIXED — sync session、install token、源路径、refresh、schedule、primary lock | +| H2 | P1/P2 batch2 | FIXED — webssh tv、retry/schedule lock、agent fail-closed、bing、install status | +| H3 | P1/P2 batch3 | FIXED — script session、settings 脱敏、agent unknown server、WS 4401 | +| H4 | 风险接受 | DOC — risk-acceptance-single-operator.md | + +## Closure + +| H | 判定 | 依据 | +|---|------|------| +| H1-H3 | FIXED | docs/changelog/2026-06-04-code-review-*.md | +| H4 | ACCEPTED | 单人运维 WS ADR-011 + Agent global legacy | + +## 改动文件清单（Gate 7） + +### 后端 +- sync_engine_v2.py +- posix_paths.py +- install.py +- auth_service.py +- sync_v2.py +- main.py +- schedule_runner.py +- retry_runner.py +- agent.py +- settings.py +- websocket.py +- webssh.py +- script_service.py + +### 前端 +- LoginPage.vue +- usePushProgress.ts +- useServerList.ts +- CommandsPage.vue +- CredentialsPage.vue +- DashboardPage.vue +- ScriptsPage.vue +- SettingsPage.vue +- api.ts + +### 静态 / 测试 +- install.html +- test_security_unit.py +- test_retry_runner_outcome.py + +### 部署 +- deploy-production.sh + +## DoD + +- [x] ruff / pytest 28 passed（.venv） +- [x] vue-tsc 通过 +- [x] changelog + audit 齐全 +- [ ] 生产 /health + /app/ 200（部署后） diff --git a/docs/changelog/2026-06-04-code-review-p0-p1-fixes.md b/docs/changelog/2026-06-04-code-review-p0-p1-fixes.md new file mode 100644 index 00000000..f948e519 --- /dev/null +++ b/docs/changelog/2026-06-04-code-review-p0-p1-fixes.md @@ -0,0 +1,43 @@ +# 2026-06-04 — Code Review P0/P1 修复批次 + +## 摘要 + +依据 Cursor 全量 code review 结论，修复安装窗口、并行推送 DB 会话、源路径校验、Refresh 轮换竞态、定时任务失败判定、Primary 锁降级及前端推送/登录问题。 + +## 动机 + +- P0：`sync_files` 并行 worker 共用 `AsyncSession` 导致间歇性 DB 错误；`create-admin` 在 Step3~4 之间无鉴权可被抢管。 +- P1：推送源路径未统一黑名单；Refresh 非原子轮换；schedule 全失败仍提交 `last_run_at`；Redis 不可用时多 worker 重复后台任务；前端 WS 泄漏与开放重定向。 + +## 涉及文件 + +| 文件 | 变更 | +|------|------| +| `server/application/services/sync_engine_v2.py` | `_db_lock` 串行化 DB 写；源路径走 `resolve_nexus_push_source_path` | +| `server/utils/posix_paths.py` | 新增 `FORBIDDEN_NEXUS_SOURCE_PREFIXES`、`resolve_nexus_push_source_path` | +| `server/api/install.py` | Step3 签发 `install_token`；`create-admin` 校验令牌 | +| `web/app/install.html` | 保存并提交 `install_token` | +| `server/application/services/auth_service.py` | Redis Lua 原子 refresh 轮换 | +| `server/api/sync_v2.py` | validate/verify 复用统一源路径校验 | +| `server/main.py` | Redis 失败时 flock 文件锁 fallback（单 primary） | +| `server/background/schedule_runner.py` | 推送零成功时 rollback schedule | +| `frontend/src/pages/LoginPage.vue` | 登录 redirect 防开放重定向 | +| `frontend/src/composables/push/usePushProgress.ts` | 失败时 disconnect WS；4001/4401 强制登出 | +| `tests/test_security_unit.py` | install token + 源路径单元测试 | + +## 迁移 / 重启 + +- 需重启 `nexus`（Supervisor）使后端生效。 +- 前端需 `vite build` 后部署静态资源。 +- **安装向导**：Step3 响应新增 `install_token`，Step4 必须携带；进行中的安装需从 Step3 重新执行以获取令牌。 + +## 验证 + +```bash +.venv/bin/ruff check server/... +.venv/bin/python -c "import server.main" +.venv/bin/pytest tests/test_security_unit.py -q +cd frontend && npm run type-check +``` + +进度：`☑实现 ☑WSL验证 □审计8步 □部署 □健康检查 □浏览器验证 ☑changelog` diff --git a/docs/changelog/2026-06-04-code-review-p1-p2-batch2.md b/docs/changelog/2026-06-04-code-review-p1-p2-batch2.md new file mode 100644 index 00000000..c0e2bb10 --- /dev/null +++ b/docs/changelog/2026-06-04-code-review-p1-p2-batch2.md @@ -0,0 +1,43 @@ +# 2026-06-04 — Code Review P1/P2 修复批次 2 + +## 摘要 + +接续 P0/P1 批次，修复 WebSSH `tv` 校验、retry/schedule 行级锁、Agent fail-closed、Bing 壁纸 DoS、安装 status 隐藏及前端静默错误。 + +## 动机 + +全量 review 剩余项：WebSSH 与 access token 策略不一致；多 worker 重复 retry/schedule；Agent DB 异常误回退全局 Key；公开 Bing API 可被滥用拉取外网；多页面加载失败无提示。 + +## 涉及文件 + +| 区域 | 文件 | +|------|------| +| 认证 | `auth_service.py`（WebSSH JWT 含 `tv`）、`webssh.py`（校验 `tv`） | +| Agent | `agent.py`（DB 查服失败 reject） | +| 壁纸 | `settings.py`（GET 只读缓存 + POST sync 需 JWT） | +| 安装 | `install.py`（锁定后 `/status` 404） | +| 后台 | `retry_runner.py`（SKIP LOCKED + running→pending）、`schedule_runner.py`（FOR UPDATE + `_schedule_is_due`） | +| 前端 | `useServerList.ts`、`CredentialsPage.vue`、`ScriptsPage.vue`、`CommandsPage.vue`、`DashboardPage.vue` | +| 测试 | `test_retry_runner_outcome.py` | + +## 行为变更 + +- **登录页壁纸**：`GET /settings/bing-wallpapers` 仅返回本地缓存；管理员登录后 Dashboard 调用 `POST /settings/bing-wallpapers/sync` 更新。 +- **WebSSH**：新签发的 token 含 `tv`；旧 token 仍走 `updated_at` 宽限。 +- **安装完成**：`GET /api/install/status` 在锁定后返回 404。 + +## 迁移 / 重启 + +- 重启 `nexus` Supervisor 服务。 +- 前端重新 `vite build` 部署。 + +## 验证 + +```bash +.venv/bin/ruff check server/... +.venv/bin/python -c "import server.main" +.venv/bin/pytest tests/test_security_unit.py tests/test_retry_runner_outcome.py -q +cd frontend && npm run type-check +``` + +进度：`☑实现 ☑WSL验证 ☑审计8步 □部署 □健康检查 □浏览器验证 ☑changelog` diff --git a/docs/changelog/2026-06-04-code-review-p1-p2-batch3.md b/docs/changelog/2026-06-04-code-review-p1-p2-batch3.md new file mode 100644 index 00000000..c758651a --- /dev/null +++ b/docs/changelog/2026-06-04-code-review-p1-p2-batch3.md @@ -0,0 +1,50 @@ +# 2026-06-04 — Code Review P1/P2 修复批次 3 + +## 摘要 + +接续批次 2，修复脚本并行 SSH 共用 AsyncSession、Settings 敏感项 API 脱敏、Telegram Token 前端留空不覆盖、Agent 未知 server 不再回退全局 Key、WebSocket/WebSSH 无效 JWT 关闭码统一为 4401。 + +## 动机 + +全量 review 剩余项：脚本批量下发并行 `get_by_id` 竞态；`GET /settings/` 仍返回 Telegram token 明文前缀；空字符串 PUT 可能误清空 token；未知 server_id 用 global key 通过认证；无效 JWT 与缺失 token 关闭码混用 4001。 + +## 涉及文件 + +| 区域 | 文件 | +|------|------| +| 脚本 | `script_service.py`（预加载 server_map，并行 SSH 前隔离 DB） | +| 设置 API | `settings.py`（`_format_setting_response`、`value_set`、敏感项空 PUT 跳过） | +| Agent | `agent.py`（server 必须存在；unknown id 拒绝；legacy global 仅无 per-server key） | +| WebSocket | `websocket.py`、`webssh.py`（无效 JWT → 4401，缺失 token 仍 4001） | +| 前端 | `SettingsPage.vue`、`types/api.ts`（Telegram token draft + reveal + value_set） | + +## 行为变更 + +- **Settings 列表/单项**：`telegram_bot_token` 等敏感项返回 `value=""` + `value_set: true`，不再带部分明文。 +- **Settings PUT**：敏感 key 提交空字符串时保留原值（未配置则 400）。 +- **Settings 页**：Bot Token 留空不提交；眼睛图标经 `/settings/telegram/reveal-token` 二次验证后填入 draft。 +- **Agent**：不存在的服务器 ID 一律 401；DB 查服异常 fail-closed。 +- **WS/WebSSH**：过期/无效 JWT 关闭码 4401，便于前端区分「未登录」与「会话失效」。 + +## 未纳入本批次（需协议变更） + +- WebSocket JWT 移出 query 字符串（需前后端首帧鉴权协议）。 +- 完全禁用 Agent global API_KEY 回退（会破坏未分配 per-server key 的旧 Agent）。 + +## 迁移 / 重启 + +- 重启 `nexus` Supervisor 服务。 +- 前端重新 `vite build` 部署。 + +## 验证 + +```bash +.venv/bin/ruff check server/application/services/script_service.py server/api/settings.py server/api/agent.py server/api/websocket.py server/api/webssh.py +.venv/bin/python -c "import server.main" +.venv/bin/pytest tests/test_security_unit.py tests/test_retry_runner_outcome.py -q +cd frontend && npm run type-check +``` + +结果：ruff ✅ · import ✅ · pytest 28 passed ✅ · vue-tsc ✅ + +进度：`☑实现 ☑WSL验证 ☑审计8步 □部署 □健康检查 □浏览器验证 ☑changelog` diff --git a/docs/changelog/2026-06-04-risk-acceptance-single-operator.md b/docs/changelog/2026-06-04-risk-acceptance-single-operator.md new file mode 100644 index 00000000..96c8e4c2 --- /dev/null +++ b/docs/changelog/2026-06-04-risk-acceptance-single-operator.md @@ -0,0 +1,32 @@ +# 2026-06-04 — 单人运维风险接受记录 + +## 摘要 + +新增 `docs/project/risk-acceptance-single-operator.md`，正式记录 Code Review 批次 3 后两项「刻意未纳入」决定在单人自用后台场景下的风险接受与重评条件；并在 AI 接续文档中建立引用。 + +## 动机 + +全量 Code Review 将 WebSocket JWT query（ADR-011）与 Agent global API_KEY legacy 回退标为需更大改动项。Operator 确认后台仅一人使用、管理员完全可信，经讨论决定不排 WS 协议改造与 global 硬禁用，需留档以免后续会话误抬为 P0。 + +## 涉及文件 + +| 文件 | 变更 | +|------|------| +| `docs/project/risk-acceptance-single-operator.md` | 新建：威胁模型、两项接受理由/残余风险/重评触发 | +| `docs/project/AI-HANDOFF-2026-06-03.md` | 增补 Code Review 收尾与风险接受链接 | +| `docs/project/AI-HANDOFF-2026-05-23.md` | §7b、§10 索引、§0 提示词、§13 时间线 | + +## 决定摘要 + +- **ADR-011（WS `?token=`）**：维持，不排首帧鉴权/ticket 改造。 +- **Agent global legacy 回退**：保留；未知 server_id 与 fail-closed 已在 batch3 收紧。 + +## 迁移 / 重启 + +无。纯文档。 + +## 验证 + +- 文档路径可访问、handoff 交叉引用一致。 + +进度：`☑实现 ☑WSL验证 □审计8步 □部署 □健康检查 □浏览器验证 ☑changelog` diff --git a/docs/project/AI-HANDOFF-2026-05-23.md b/docs/project/AI-HANDOFF-2026-05-23.md index 28f4dde6..df4d84a9 100644 --- a/docs/project/AI-HANDOFF-2026-05-23.md +++ b/docs/project/AI-HANDOFF-2026-05-23.md @@ -37,7 +37,7 @@ 项目未部署。端口：子机出站443心跳；中心→子机SSH22；8601仅本机。 P1待办：WebSSH断线重连、CSV批量导入。首次部署E2E见本文§6。 -已否决见§7。 +已否决见§7。风险接受（单人运维·不排期）见§7b。 ``` --- @@ -232,6 +232,22 @@ Nexus 6.0：2000+ 台服务器运维平台；FastAPI 四层架构 + Redis 实时 --- +## 7b. 风险接受（单人运维 · 不排期，2026-06-04） + +> **SSOT**: `docs/project/risk-acceptance-single-operator.md` +> Operator 确认：后台仅 **一名** 操作员自用，管理员身份完全可信。 + +| 项 | 决定 | 说明 | +|----|------|------| +| WebSocket JWT query | **维持 ADR-011** | 不排首帧鉴权 / WS ticket；batch3 已统一 4401 | +| Agent global API_KEY 回退 | **保留 legacy** | 不硬禁用；batch3 已禁 unknown server_id + fail-closed | + +**≠ 已否决**：若新增管理员、日志外发第三方、等保审计、global key 泄露或 Agent 全量 per-server 迁移，须重评并见 risk-acceptance 文档「重评触发条件」。 + +Code Review 三批 changelog：`docs/changelog/2026-06-04-code-review-*.md` + +--- + ## 8. 文档债务（下一任 AI 应同步） | 文档 | 问题 | 动作 | @@ -279,6 +295,8 @@ Nexus 6.0：2000+ 台服务器运维平台；FastAPI 四层架构 + Redis 实时 | 健康 SSH | `server/infrastructure/ssh/remote_probe.py` | | 告警历史 | `web/app/alerts.html`, `server/api/settings.py`（alert_history_router） | | Changelog 示例 | `docs/changelog/2026-05-23-ssh-exec-no-8601.md` | +| 风险接受（单人运维） | `docs/project/risk-acceptance-single-operator.md` | +| Code Review 2026-06-04 | `docs/changelog/2026-06-04-code-review-p0-p1-fixes.md` 等三批 | --- @@ -312,6 +330,7 @@ web/app/*.html → server/api/*.py → application/services → infrastructure/* 6. Agent 8601 vs SSH → **脚本+健康检查走 22；心跳走 443；弱化 8601** 7. 宝塔 Python 管理 → **不用**；venv + 3.12 8. 项目未部署 → **无需重启** +9. 2026-06-04 Code Review 三批本地验证；单人运维风险接受 → `risk-acceptance-single-operator.md` §7b --- diff --git a/docs/project/AI-HANDOFF-2026-06-03.md b/docs/project/AI-HANDOFF-2026-06-03.md index 6db0f846..0875245c 100644 --- a/docs/project/AI-HANDOFF-2026-06-03.md +++ b/docs/project/AI-HANDOFF-2026-06-03.md @@ -25,6 +25,8 @@ 【规范】.cursor/rules/perfect-implementation.mdc · 单 Agent · 改代码必 changelog 【文档】docs/project/AI-HANDOFF-2026-06-03.md · linux-dev-paths.md · docker/README.md +【Code Review】批次 1～3 已本地验证；见 docs/changelog/2026-06-04-code-review-*.md +【风险接受 · 单人运维】docs/project/risk-acceptance-single-operator.md（WS ADR-011 + Agent global 不排期） 【插件】docs/project/cursor-plugins-for-nexus.md · MySQL MCP: scripts/linux_mcp_mysql.sh 【本地验收】seed_local_admin.py + frontend npm run test:e2e（见 linux-dev-paths.md） ``` @@ -36,3 +38,22 @@ export NEXUS_ROOT=~/Nexus mkdir -p "$NEXUS_ROOT" && tar xzf nexus-handoff-2026-06-03.tar.gz -C "$(dirname "$NEXUS_ROOT")" cd "$NEXUS_ROOT" && git remote -v # 若无 .git：重新 clone 后覆盖源码目录 ``` + +--- + +## Code Review 收尾（2026-06-04） + +本地已修复并验证三批（P0/P1 + batch2 + batch3），changelog/audit 见： + +| 批次 | Changelog | Audit | +|------|-----------|-------| +| P0/P1 | `docs/changelog/2026-06-04-code-review-p0-p1-fixes.md` | `docs/audit/2026-06-04-code-review-p0-p1-fixes.md` | +| batch2 | `docs/changelog/2026-06-04-code-review-p1-p2-batch2.md` | `docs/audit/2026-06-04-code-review-p1-p2-batch2.md` | +| batch3 | `docs/changelog/2026-06-04-code-review-p1-p2-batch3.md` | `docs/audit/2026-06-04-code-review-p1-p2-batch3.md` | + +**未排期（已风险接受）**：Operator 确认后台仅一人自用 → 见 **`docs/project/risk-acceptance-single-operator.md`** + +- WebSocket JWT 仍走 query（ADR-011），不改造协议 +- Agent global `API_KEY` legacy 回退保留；batch3 已禁 unknown server_id + +重评触发：新增管理员、日志外发、等保审计、global key 泄露或 Agent 全量 per-server 迁移完成。 diff --git a/docs/project/risk-acceptance-single-operator.md b/docs/project/risk-acceptance-single-operator.md new file mode 100644 index 00000000..c5386ba6 --- /dev/null +++ b/docs/project/risk-acceptance-single-operator.md @@ -0,0 +1,126 @@ +# 风险接受记录 — 单人运维场景（2026-06-04） + +## 背景 + +Nexus 全量 Code Review（批次 1～3）完成后，有两项被标记为「刻意未纳入」，需更大协议变更或 Agent 全站迁移： + +1. WebSocket JWT 仍通过 query 传递（ADR-011） +2. Agent 认证保留 global `API_KEY` 对未分配 `agent_api_key` 服务器的 legacy 回退 + +**运营事实（2026-06-04 确认）**：后台仅 **一名** 操作员自用，无多人共享账号、无外包运维、无多租户。操作员即系统 owner，管理员身份视为完全可信。 + +本文档记录在该威胁模型下的**正式风险接受**与重评条件，避免后续审计或接续开发误将两项抬为 P0。 + +--- + +## 威胁模型（单人运维） + +| 维度 | 假设 | +|------|------| +| 后台用户 | 仅 1 人，可信 | +| 内鬼 / 权限滥用 | 不适用 | +| 主要防护目标 | 外网未授权访问、安装/认证漏洞、多 worker 竞态、子机 Agent M2M 密钥管理 | +| 日志访问 | Nginx/宝塔等 access log 仅 operator 可阅，不外发第三方 | + +--- + +## 接受项 1：WebSocket JWT query（ADR-011） + +### 现状 + +- `/ws/alerts`、`/ws/sync`、`/ws/terminal/{id}` 均在握手 URL 使用 `?token=...` +- 设计依据：ADR-011；浏览器 WebSocket 无法自定义 Authorization Header +- 批次 3 已统一：缺失 token → **4001**；无效/过期 JWT → **4401** + +### 接受理由 + +- Token 仅在 operator 登录后由其浏览器持有，无「不可信管理员」场景 +- 改首帧鉴权 / WS ticket / 子协议需前后端同发、三条通道联测，**单人自用 ROI 极低** +- 残余风险（access log 留存 JWT、DevTools 可见）在「仅 operator 可读日志」前提下可接受 + +### 残余风险 + +- 本机恶意软件可窃取 session（与 token 在 query 或 Header 无本质差别） +- 若未来日志外发 SIEM/第三方，URL 中的 JWT 成为额外泄露面 + +### 重评触发条件 + +- 新增第二管理员或外包运维 +- 等保/客户安全审计要求「凭证不得出现在 URL」 +- access log 外发或多人可读 +- 后台改为多租户或 SaaS 形态 + +### 决定 + +**维持 ADR-011，不排 WS 协议改造。** + +--- + +## 接受项 2：Agent global API_KEY legacy 回退 + +### 现状 + +- 子机 Agent 通过 `X-API-Key` 调用 `/api/agent/heartbeat`（无 JWT、无后台登录） +- 认证逻辑（`server/api/agent.py`）： + 1. 服务器必须存在 + 2. 若已分配 `agent_api_key` → 必须匹配 per-server key + 3. **Legacy**：未分配 per-server key → 仍接受 global `API_KEY`（写 warning 日志） +- 批次 3 已收紧：未知 `server_id` 拒绝；DB 查服异常 fail-closed + +### 接受理由 + +- 与「后台几人用」无关，但单人运维下 global key 分发面可控（`.env`、安装脚本、子机 `config.json`） +- 硬禁用 global 回退可能导致大量 legacy Agent 心跳 401，**可用性风险大于当前威胁** +- 已有 per-server key 的服务器已与 global 隔离 + +### 残余风险 + +- global `API_KEY` 若从备份、误提交、子机配置泄露，攻击者可对 **尚未分配 `agent_api_key`** 的服务器伪造心跳（污染在线状态/监控数据，非 RCE） +- 攻击者需同时能访问中心 API 地址并持有 key + +### 缓解（已实施） + +- 未知 server_id 不再接受 global 冒充 +- 新装/生成 key 流程优先 per-server（`POST /servers/{id}/agent-key`） +- legacy 使用 global 时服务器日志告警，便于逐步迁移 + +### 重评触发条件 + +- global key 疑似或确认泄露 +- 计划轮换 `API_KEY` +- 新装 Agent 100% per-server 且存量迁移完成 +- 监控/告警数据完整性成为合规要求 + +### 决定 + +**保留 legacy global 回退；不排硬禁用。** 有空时可做摸底（`agent_api_key IS NULL` 数量）与渐进迁移，非阻塞项。 + +--- + +## 仍须保持的修复（与单人场景无关） + +以下项已在 Code Review 批次 1～3 修复，属于**稳定性/外网暴露**问题，不因单人运维而降级： + +- Sync/脚本并行任务 AsyncSession 竞态 +- 安装向导 create-admin 鉴权窗口 +- schedule/retry 多 worker 重复执行 +- 登录开放重定向、推送 WS 断线与 logout 码 +- Settings 敏感项脱敏、Agent unknown server 拒绝等 + +--- + +## 变更记录 + +| 日期 | 说明 | +|------|------| +| 2026-06-04 | 初始记录；operator 确认单人自用后台 | + +--- + +## 引用 + +- ADR-011：`docs/project/roadmap.md`（WebSocket JWT query） +- Code Review 批次 3：`docs/changelog/2026-06-04-code-review-p1-p2-batch3.md` +- 本记录 changelog：`docs/changelog/2026-06-04-risk-acceptance-single-operator.md` +- AI 接续：`docs/project/AI-HANDOFF-2026-05-23.md` §7b · `docs/project/AI-HANDOFF-2026-06-03.md` §Code Review 收尾 +- Agent 认证实现：`server/api/agent.py` — `_verify_server_api_key` diff --git a/frontend/src/composables/push/usePushProgress.ts b/frontend/src/composables/push/usePushProgress.ts index 26691662..042b7052 100644 --- a/frontend/src/composables/push/usePushProgress.ts +++ b/frontend/src/composables/push/usePushProgress.ts @@ -162,6 +162,10 @@ export function usePushProgress( socket.onclose = (ev) => { wsSocket.value = null + if (ev.code === 4001 || ev.code === 4401) { + auth.forceLogout() + return + } if (ev.code !== 1000 && pushing.value) { wsProgressItems.value.forEach(p => { if (p.status === 'pending' || p.status === 'running') { @@ -242,6 +246,7 @@ export function usePushProgress( } applyPushResultsFromHttp(res) } catch (e: unknown) { + disconnectProgressWs() const msg = e instanceof Error ? e.message : '推送请求失败' wsProgressItems.value.forEach(p => { p.status = 'failed' @@ -249,6 +254,7 @@ export function usePushProgress( }) pushing.value = false snackbar(msg, 'error') + onPushFinished() return } diff --git a/frontend/src/composables/useServerList.ts b/frontend/src/composables/useServerList.ts index cdcfcf25..57d36e08 100644 --- a/frontend/src/composables/useServerList.ts +++ b/frontend/src/composables/useServerList.ts @@ -6,6 +6,8 @@ */ import { ref } from 'vue' import { http } from '@/api' +import { useSnackbar } from '@/composables/useSnackbar' +import { formatApiError } from '@/utils/apiError' export interface ServerBrief { id: number @@ -19,6 +21,7 @@ export interface ServerBrief { export function useServerList() { const servers = ref([]) const loading = ref(false) + const snackbar = useSnackbar() async function loadServers(opts?: { perPage?: number }) { loading.value = true @@ -27,8 +30,9 @@ export function useServerList() { per_page: opts?.perPage ?? 200, }) servers.value = res.items - } catch { + } catch (e: unknown) { servers.value = [] + snackbar(formatApiError(e, '加载服务器列表失败'), 'error') } finally { loading.value = false } diff --git a/frontend/src/pages/CommandsPage.vue b/frontend/src/pages/CommandsPage.vue index b32c4edd..17958835 100644 --- a/frontend/src/pages/CommandsPage.vue +++ b/frontend/src/pages/CommandsPage.vue @@ -99,8 +99,11 @@ import { ref, onMounted } from 'vue' import { http } from '@/api' import { useServerList } from '@/composables/useServerList' +import { useSnackbar } from '@/composables/useSnackbar' +import { formatApiError } from '@/utils/apiError' import type { CommandLogItem, SshSessionItem } from '@/types/api' +const snackbar = useSnackbar() const { servers: serverList, loadServers } = useServerList() // ── State ── @@ -149,9 +152,10 @@ async function loadData() { sessions.value = res.items total.value = res.total } - } catch { + } catch (e: unknown) { commands.value = [] sessions.value = [] + snackbar(formatApiError(e, '加载命令日志失败'), 'error') } finally { loading.value = false } diff --git a/frontend/src/pages/CredentialsPage.vue b/frontend/src/pages/CredentialsPage.vue index 0abb1c70..8393604b 100644 --- a/frontend/src/pages/CredentialsPage.vue +++ b/frontend/src/pages/CredentialsPage.vue @@ -182,6 +182,7 @@ import { ref, watch, onMounted } from 'vue' import { http } from '@/api' import { useSnackbar } from '@/composables/useSnackbar' +import { formatApiError } from '@/utils/apiError' import { required } from '@/utils/validation' const snackbar = useSnackbar() @@ -252,7 +253,10 @@ async function loadPasswords() { const res = await http.getList('/presets/') passwords.value = res.items totalItems.value = res.total - } catch { passwords.value = [] } + } catch (e: unknown) { + passwords.value = [] + snackbar(formatApiError(e, '加载密码预设失败'), 'error') + } finally { loading.value = false } } async function loadSshKeys() { @@ -264,7 +268,10 @@ async function loadSshKeys() { }) sshKeys.value = res.items totalItems.value = res.total - } catch { sshKeys.value = [] } + } catch (e: unknown) { + sshKeys.value = [] + snackbar(formatApiError(e, '加载 SSH 密钥失败'), 'error') + } finally { loading.value = false } } async function loadDbCreds() { @@ -276,7 +283,10 @@ async function loadDbCreds() { }) dbCreds.value = res.items totalItems.value = res.total - } catch { dbCreds.value = [] } + } catch (e: unknown) { + dbCreds.value = [] + snackbar(formatApiError(e, '加载数据库凭据失败'), 'error') + } finally { loading.value = false } } diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index e0c165c6..43e65979 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -332,6 +332,7 @@ async function loadAlerts() { })) } catch { recentAlerts.value = [] + snackbar('加载告警记录失败', 'error') } } @@ -339,7 +340,10 @@ async function loadRecentAudit() { try { const res = await http.get<{ items: unknown[] }>('/audit/', { limit: 10 }) recentAudit.value = normalizeAuditList(res.items || []).slice(0, 10) - } catch { recentAudit.value = [] } + } catch { + recentAudit.value = [] + snackbar('加载审计记录失败', 'error') + } } async function loadSummary() { @@ -398,6 +402,7 @@ onMounted(() => { loadAlerts() loadSummary() loadRecentAudit() + http.post('/settings/bing-wallpapers/sync').catch(() => {}) }) interface AlertItem { diff --git a/frontend/src/pages/LoginPage.vue b/frontend/src/pages/LoginPage.vue index de9a51cd..82fe3fb7 100644 --- a/frontend/src/pages/LoginPage.vue +++ b/frontend/src/pages/LoginPage.vue @@ -175,7 +175,8 @@ async function doLogin() { try { await auth.login(username.value, password.value, needTotp.value ? totpCode.value : undefined) - const redirect = (route.query.redirect as string) || '/' + const raw = (route.query.redirect as string) || '/' + const redirect = raw.startsWith('/') && !raw.startsWith('//') ? raw : '/' router.push(redirect) } catch (e: any) { if (e instanceof TotpRequiredError) { diff --git a/frontend/src/pages/ScriptsPage.vue b/frontend/src/pages/ScriptsPage.vue index b1af7f9a..9a6eed70 100644 --- a/frontend/src/pages/ScriptsPage.vue +++ b/frontend/src/pages/ScriptsPage.vue @@ -260,6 +260,7 @@ import { ref, computed, watch, onMounted, onUnmounted } from 'vue' import { http } from '@/api' import { useServerList } from '@/composables/useServerList' import { useSnackbar } from '@/composables/useSnackbar' +import { formatApiError } from '@/utils/apiError' import { required } from '@/utils/validation' const snackbar = useSnackbar() @@ -360,7 +361,10 @@ async function loadScripts() { }) scripts.value = res.items totalItems.value = res.total - } catch { scripts.value = [] } + } catch (e: unknown) { + scripts.value = [] + snackbar(formatApiError(e, '加载脚本列表失败'), 'error') + } finally { loading.value = false } } diff --git a/frontend/src/pages/SettingsPage.vue b/frontend/src/pages/SettingsPage.vue index de94cda8..f3ed8810 100644 --- a/frontend/src/pages/SettingsPage.vue +++ b/frontend/src/pages/SettingsPage.vue @@ -55,7 +55,18 @@ - + + Token 已配置 发送测试消息 @@ -162,15 +173,17 @@ - 验证身份 + {{ revealTarget === 'telegram' ? '查看 Bot Token' : '验证身份' }} -
请输入当前密码以查看 API Key
+
+ {{ revealTarget === 'telegram' ? '请输入当前密码以查看 Telegram Bot Token' : '请输入当前密码以查看 API Key' }} +
取消 - 确认 + 确认 @@ -214,6 +227,8 @@ const settings = ref({ db_pool_size: 10, db_max_overflow: 20, telegram_bot_token: '', telegram_chat_id: '', }) +const telegramTokenDraft = ref('') +const telegramTokenSet = ref(false) const saving = ref(false) const testingTg = ref(false) const showToken = ref(false) @@ -236,6 +251,7 @@ const disablingTotp = ref(false) const showApiKey = ref(false) const apiKeyValue = ref('') const showRevealDialog = ref(false) +const revealTarget = ref<'api_key' | 'telegram'>('api_key') const revealPassword = ref('') const revealingKey = ref(false) @@ -253,6 +269,11 @@ async function loadSettings() { const knownKeys = Object.keys(settings.value) for (const item of items) { const key = item.key + if (key === 'telegram_bot_token') { + telegramTokenSet.value = Boolean(item.value_set) + telegramTokenDraft.value = '' + continue + } if (key && knownKeys.includes(key)) { const val = item.value // Type-cast numeric strings back to numbers for number inputs @@ -283,14 +304,23 @@ async function loadAllowlist() { // ── Actions ── async function saveSettings() { saving.value = true + const tokenUpdate = telegramTokenDraft.value.trim() try { const entries = Object.entries(settings.value).filter(([key]) => - ['system_name', 'system_title', 'cpu_alert_threshold', 'mem_alert_threshold', 'disk_alert_threshold', 'db_pool_size', 'db_max_overflow', 'telegram_bot_token', 'telegram_chat_id'].includes(key) + ['system_name', 'system_title', 'cpu_alert_threshold', 'mem_alert_threshold', 'disk_alert_threshold', 'db_pool_size', 'db_max_overflow', 'telegram_chat_id'].includes(key) ) + if (tokenUpdate) { + entries.push(['telegram_bot_token', tokenUpdate]) + } for (const [key, value] of entries) { await http.put(`/settings/${key}`, { value }) } snackbar('设置已保存') + if (tokenUpdate) { + telegramTokenSet.value = true + telegramTokenDraft.value = '' + showToken.value = false + } } catch (e: unknown) { const msg = e instanceof Error ? e.message : '保存失败' snackbar(msg, 'error') @@ -381,25 +411,58 @@ function copyTotpSecret() { } async function revealApiKey() { - if (showApiKey.value) { showApiKey.value = false; return } + if (showApiKey.value) { + showApiKey.value = false + return + } + revealTarget.value = 'api_key' showRevealDialog.value = true revealPassword.value = '' } -async function doRevealApiKey() { +async function revealTelegramToken() { + if (showToken.value && telegramTokenDraft.value) { + showToken.value = false + telegramTokenDraft.value = '' + return + } + revealTarget.value = 'telegram' + showRevealDialog.value = true + revealPassword.value = '' +} + +async function doRevealSecret() { if (!revealPassword.value) { snackbar('请输入密码', 'error') return } revealingKey.value = true try { - const res = await http.post<{ api_key: string }>('/settings/api-key/reveal', { current_password: revealPassword.value }) - apiKeyValue.value = res.api_key - showApiKey.value = true + if (revealTarget.value === 'telegram') { + const res = await http.post<{ value: string }>('/settings/telegram/reveal-token', { + current_password: revealPassword.value, + }) + telegramTokenDraft.value = res.value + showToken.value = true + } else { + const res = await http.post<{ api_key: string }>('/settings/api-key/reveal', { + current_password: revealPassword.value, + }) + apiKeyValue.value = res.api_key + showApiKey.value = true + } showRevealDialog.value = false revealPassword.value = '' - } catch (e: any) { snackbar(e.message || '验证失败', 'error') } - finally { revealingKey.value = false } + } catch (e: unknown) { + snackbar(e instanceof Error ? e.message : '验证失败', 'error') + } finally { + revealingKey.value = false + } +} + +async function doRevealApiKey() { + revealTarget.value = 'api_key' + await doRevealSecret() } function copyApiKey() { diff --git a/frontend/src/types/api.ts b/frontend/src/types/api.ts index a9261792..47289801 100644 --- a/frontend/src/types/api.ts +++ b/frontend/src/types/api.ts @@ -18,6 +18,7 @@ export type SettingsResponse = SettingItem[] export interface SettingItem { key: string value: unknown + value_set?: boolean | null updated_at?: string } diff --git a/server/api/agent.py b/server/api/agent.py index 08edbd84..3e6cb218 100644 --- a/server/api/agent.py +++ b/server/api/agent.py @@ -58,22 +58,39 @@ def _verify_api_key(x_api_key: str = Header(...)): async def _verify_server_api_key(server_id: int, provided_key: str, service: ServerService) -> bool: - """Verify API key against per-server agent_api_key, falling back to global API key. + """Verify API key against per-server agent_api_key, with limited global fallback. Authentication priority: - 1. If server has agent_api_key set → must match exactly - 2. Otherwise → must match global API_KEY + 1. Server must exist + 2. If server has agent_api_key → must match exactly + 3. Legacy only: no per-server key → global API_KEY (logged for migration) """ - # Try to get server's per-server key try: server = await service.get_server(server_id) - if server and server.agent_api_key: - return secrets.compare_digest(provided_key, server.agent_api_key) except Exception: - logger.debug(f"Failed to look up server {server_id} for per-server API key check, falling back to global key") + logger.warning( + "Failed to look up server %s for per-server API key check — rejecting", + server_id, + exc_info=True, + ) + return False - # Fallback: global API key - return secrets.compare_digest(provided_key, settings.API_KEY) + if not server: + return False + + if server.agent_api_key: + return secrets.compare_digest(provided_key, server.agent_api_key) + + if secrets.compare_digest(provided_key, settings.API_KEY): + logger.warning( + "Server %s (%s) used global API_KEY for agent auth — " + "assign per-server key via POST /api/servers/%s/agent-key", + server_id, + server.name, + server_id, + ) + return True + return False def _threshold_for(thresholds: dict, metric: str) -> float: diff --git a/server/api/install.py b/server/api/install.py index 9e05a0ad..7d79e441 100644 --- a/server/api/install.py +++ b/server/api/install.py @@ -68,6 +68,7 @@ class InitDbRequest(BaseModel): class CreateAdminRequest(BaseModel): + install_token: str db_host: str = "localhost" db_port: str = "3306" db_name: str = "Nexus" @@ -310,11 +311,29 @@ def _finalize_install_lock() -> None: logger.warning("Failed to remove install state file: %s", e) +def _verify_install_token(provided: str) -> None: + """Require one-time install token issued at step 3 (init-db).""" + if not provided or not provided.strip(): + raise HTTPException(status_code=403, detail="缺少安装令牌，请从步骤3重新开始") + if not STATE_FILE.exists(): + raise HTTPException(status_code=403, detail="安装会话无效，请从步骤3重新开始") + import json + try: + state = json.loads(read_utf8_text(STATE_FILE)) + except (json.JSONDecodeError, OSError) as exc: + raise HTTPException(status_code=403, detail="安装会话无效，请从步骤3重新开始") from exc + expected = state.get("install_token") + if not expected or not secrets.compare_digest(provided.strip(), expected): + raise HTTPException(status_code=403, detail="安装令牌无效或已过期") + + # ── Endpoints ── @router.get("/status") async def install_status(): """Check whether Nexus is already installed.""" + if _is_locked(): + raise HTTPException(status_code=404, detail="Not found") return { "installed": _is_installed(), "locked": _is_locked(), @@ -542,6 +561,7 @@ async def init_db(req: InitDbRequest): # Save install state for step 4 import json + install_token = secrets.token_urlsafe(32) state = { "db_host": req.db_host, "db_port": req.db_port, @@ -549,6 +569,7 @@ async def init_db(req: InitDbRequest): "db_user": req.db_user, # P2-19: Don't store db_pass in state file — only needed for _make_db_url() # which is called with the full CreateAdminRequest in step 4 + "install_token": install_token, "pool_size": pool_size, "max_overflow": max_overflow, "install_dir": install_dir, @@ -561,6 +582,7 @@ async def init_db(req: InitDbRequest): return { "success": True, + "install_token": install_token, "pool_size": pool_size, "max_overflow": max_overflow, "tables_created": 14, @@ -576,6 +598,8 @@ async def create_admin(req: CreateAdminRequest): if not _has_env(): raise HTTPException(400, "请先完成步骤3：数据库初始化") + _verify_install_token(req.install_token) + if len(req.admin_password) < 6: raise HTTPException(400, "密码长度至少6位") diff --git a/server/api/settings.py b/server/api/settings.py index 2f332a6e..5a40cde8 100644 --- a/server/api/settings.py +++ b/server/api/settings.py @@ -111,18 +111,29 @@ def _validate_ip_list(ips: list[str]) -> None: # ── System Settings ── +def _format_setting_response(key: str, value: str | None, updated_at) -> dict: + """Build setting dict; sensitive keys never return plaintext.""" + if key in SENSITIVE_KEYS and value: + return { + "key": key, + "value": "", + "value_set": True, + "updated_at": str(updated_at), + } + return { + "key": key, + "value": value, + "value_set": bool(value) if key in SENSITIVE_KEYS else None, + "updated_at": str(updated_at), + } + + @router.get("/", response_model=list) async def list_settings(admin: Admin = Depends(get_current_admin), db: AsyncSession = Depends(get_db)): """List all system settings (sensitive values masked)""" repo = SettingRepositoryImpl(db) settings = await repo.get_all() - result = [] - for s in settings: - val = s.value - if s.key in SENSITIVE_KEYS and val: - val = val[:8] + "..." if len(val) > 8 else "***" - result.append({"key": s.key, "value": val, "updated_at": str(s.updated_at)}) - return result + return [_format_setting_response(s.key, s.value, s.updated_at) for s in settings] @router.get("/ip-allowlist", response_model=dict) @@ -173,22 +184,29 @@ def _cleanup_old_wallpapers() -> int: return removed -@router.get("/bing-wallpapers", response_model=dict) -async def bing_wallpapers(): - """Return all cached wallpaper URLs for frontend slideshow rotation. +def _cached_wallpaper_urls() -> dict: + """List locally cached wallpaper URLs (no outbound HTTP).""" + _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True) + all_files = sorted( + _WALLPAPER_DIR.glob("*.jpg"), + key=lambda f: f.stat().st_mtime, + reverse=True, + ) + urls = [ + f"/app/wallpapers/{f.name}" + for f in sorted(all_files[:_MAX_WALLPAPERS], key=lambda f: f.name, reverse=True) + ] + return {"urls": urls, "count": len(urls)} - Syncs with Bing daily (pulls up to 8 days of images), caches to disk, - cleans up old files, returns sorted list of local URLs. - """ + +async def _sync_bing_wallpapers_from_network() -> int: + """Download recent Bing images to local cache. Returns count downloaded.""" import httpx _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True) - - # Download today + last 7 days from Bing (n=8 gives ~8 images) downloaded = 0 try: async with httpx.AsyncClient(timeout=12.0) as client: - # Fetch up to 8 recent images from Bing resp = await client.get( "https://cn.bing.com/HPImageArchive.aspx", params={"format": "js", "idx": 0, "n": 8, "mkt": "zh-CN"}, @@ -198,17 +216,15 @@ async def bing_wallpapers(): img_path = img.get("url") if not img_path: continue - # Use date from Bing metadata as filename date_str = img.get("startdate", "") or img.get("fullstartdate", "")[:8] or "" if not date_str: - # fallback: extract from URL import re - m = re.search(r'OHR\.(\w+)_', img_path) + m = re.search(r"OHR\.(\w+)_", img_path) date_str = m.group(1) if m else f"unknown_{downloaded}" cached = _WALLPAPER_DIR / f"{date_str}.jpg" if cached.exists() and cached.stat().st_size > 1024: - continue # already cached + continue try: img_resp = await client.get(f"https://cn.bing.com{img_path}") @@ -217,64 +233,50 @@ async def bing_wallpapers(): downloaded += 1 except httpx.HTTPError as e: logger.debug("Bing image download failed: %s", e) - continue except httpx.HTTPError as e: logger.warning("Bing wallpaper sync failed: %s", e) - # Cleanup old files _cleanup_old_wallpapers() - - # Also keep only newest N files if we somehow have too many all_files = sorted(_WALLPAPER_DIR.glob("*.jpg"), key=lambda f: f.stat().st_mtime, reverse=True) for f in all_files[_MAX_WALLPAPERS:]: try: f.unlink() except OSError as e: logger.debug("Wallpaper prune skip %s: %s", f.name, e) + return downloaded - # Return sorted URLs (newest first) - urls = [f"/app/wallpapers/{f.name}" for f in sorted(all_files[:_MAX_WALLPAPERS], key=lambda f: f.name, reverse=True)] - return {"urls": urls, "count": len(urls)} + +@router.get("/bing-wallpapers", response_model=dict) +async def bing_wallpapers(): + """Return cached wallpaper URLs (public, read-only — no outbound sync).""" + return _cached_wallpaper_urls() + + +@router.post("/bing-wallpapers/sync", response_model=dict) +async def bing_wallpapers_sync(admin: Admin = Depends(get_current_admin)): + """Admin-only: fetch latest Bing images into local cache.""" + downloaded = await _sync_bing_wallpapers_from_network() + result = _cached_wallpaper_urls() + result["downloaded"] = downloaded + return result # Keep the old single-image endpoint for backward compatibility @router.get("/bing-wallpaper", response_model=dict) async def bing_wallpaper(): - """Return today's wallpaper URL (backward compat).""" - try: - import httpx - from datetime import datetime + """Return today's cached wallpaper URL (public, read-only).""" + _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True) + from datetime import datetime - _WALLPAPER_DIR.mkdir(parents=True, exist_ok=True) - today = datetime.utcnow().strftime("%Y-%m-%d") - cached = _WALLPAPER_DIR / f"{today}.jpg" - - if cached.exists() and cached.stat().st_size > 1024: - return {"url": f"/app/wallpapers/{today}.jpg"} - - async with httpx.AsyncClient(timeout=10.0) as client: - resp = await client.get( - "https://cn.bing.com/HPImageArchive.aspx", - params={"format": "js", "idx": 0, "n": 1, "mkt": "zh-CN"}, - ) - data = resp.json() - img = (data.get("images") or [{}])[0] - img_url = img.get("url") - if not img_url: - return {"url": None} - - img_resp = await client.get(f"https://cn.bing.com{img_url}") - if img_resp.status_code == 200: - cached.write_bytes(img_resp.content) - - _cleanup_old_wallpapers() + today = datetime.utcnow().strftime("%Y-%m-%d") + cached = _WALLPAPER_DIR / f"{today}.jpg" + if cached.exists() and cached.stat().st_size > 1024: return {"url": f"/app/wallpapers/{today}.jpg"} - except Exception: - # Try any cached file - cached_list = sorted(_WALLPAPER_DIR.glob("*.jpg")) - if cached_list: - return {"url": f"/app/wallpapers/{cached_list[-1].name}"} - return {"url": None} + + all_files = sorted(_WALLPAPER_DIR.glob("*.jpg"), key=lambda f: f.stat().st_mtime, reverse=True) + if all_files: + return {"url": f"/app/wallpapers/{all_files[0].name}"} + return {"url": None} @router.get("/{key}", response_model=dict) @@ -284,9 +286,7 @@ async def get_setting(key: str, admin: Admin = Depends(get_current_admin), db: A value = await repo.get(key) if value is None: raise HTTPException(status_code=404, detail="Setting not found") - if key in SENSITIVE_KEYS and value: - value = value[:8] + "..." if len(value) > 8 else "***" - return {"key": key, "value": value} + return _format_setting_response(key, value, None) @router.post("/api-key/reveal", response_model=dict) @@ -333,8 +333,15 @@ async def set_setting( if key not in MUTABLE_KEYS: raise HTTPException(status_code=403, detail=f"未知设置项: {key}") - # S-02: Validate value type and range for int settings val_str = str(payload.value) + if key in SENSITIVE_KEYS and not val_str.strip(): + repo = SettingRepositoryImpl(db) + current = await repo.get(key) + if not current: + raise HTTPException(status_code=400, detail=f"'{key}' 不能为空") + return _format_setting_response(key, current, None) + + # S-02: Validate value type and range for int settings if key in SETTING_VALIDATORS: expected_type, min_val, max_val = SETTING_VALIDATORS[key] try: diff --git a/server/api/sync_v2.py b/server/api/sync_v2.py index 201dd7b9..3820e0de 100644 --- a/server/api/sync_v2.py +++ b/server/api/sync_v2.py @@ -44,6 +44,7 @@ from server.utils.posix_paths import ( normalize_remote_dest, remote_join, resolve_nexus_host_path, + resolve_nexus_push_source_path, to_posix, ) from server.application.services.sync_engine_v2 import SyncEngineV2 @@ -529,8 +530,6 @@ async def reconcile_stale_sync_logs( # ── H5: Validate Source Path ── -_FORBIDDEN_PATH_PREFIXES = ("/etc", "/root", "/home", "/var", "/boot", "/dev", "/proc", "/sys", "/run", "/srv") - @router.post("/validate-source-path") async def validate_source_path( @@ -545,25 +544,13 @@ async def validate_source_path( """ import os - path = payload.path.strip() - try: - real_path = resolve_nexus_host_path(path) + real_path = resolve_nexus_push_source_path(payload.path) except PosixPathError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - if ".." in to_posix(path).split("/"): - raise HTTPException(status_code=400, detail="路径不允许包含 '..'") - - # Reject sensitive paths - for prefix in _FORBIDDEN_PATH_PREFIXES: - if real_path == prefix or real_path.startswith(prefix + "/"): - raise HTTPException(status_code=403, detail=f"不允许使用系统路径: {prefix}") - - if not os.path.exists(real_path): - raise HTTPException(status_code=404, detail="路径不存在") - - if not os.path.isdir(real_path): - raise HTTPException(status_code=400, detail="路径不是目录") + status = 403 if "系统路径" in str(exc) else 400 + if "不存在" in str(exc): + status = 404 + raise HTTPException(status_code=status, detail=str(exc)) from exc # Count files and total size file_count = 0 @@ -582,7 +569,7 @@ async def validate_source_path( await _audit_sync( "validate_source_path", "sync", 0, - f"验证源路径: {path} ({file_count} 文件)", + f"验证源路径: {payload.path} ({file_count} 文件)", admin.username, request, ) @@ -1105,15 +1092,18 @@ async def verify_sync( session = request.state.db - if not os.path.isdir(payload.source_path): - raise HTTPException(status_code=400, detail=f"源路径不存在: {payload.source_path}") + try: + source_path = resolve_nexus_push_source_path(payload.source_path) + except PosixPathError as exc: + status = 403 if "系统路径" in str(exc) else 400 + raise HTTPException(status_code=status, detail=str(exc)) from exc # Build local file manifest: {relative_path: sha256hex} local_files = {} - for root, _dirs, fnames in os.walk(payload.source_path): + for root, _dirs, fnames in os.walk(source_path): for fn in fnames: full = posix_join(root, fn) - rel = posixpath.relpath(full, to_posix(payload.source_path)) + rel = posixpath.relpath(full, to_posix(source_path)) if len(local_files) >= payload.max_files: break try: diff --git a/server/api/websocket.py b/server/api/websocket.py index 33272b26..f1f8e5ad 100644 --- a/server/api/websocket.py +++ b/server/api/websocket.py @@ -307,7 +307,7 @@ async def alert_ws( admin = await _verify_ws_token(token) if not admin: - await websocket.close(code=4001, reason="Invalid or expired JWT token") + await websocket.close(code=4401, reason="Invalid or expired JWT token") return # ── Accept and register ── @@ -343,7 +343,7 @@ async def sync_progress_ws( admin = await _verify_ws_token(token) if not admin: - await websocket.close(code=4001, reason="Invalid or expired JWT token") + await websocket.close(code=4401, reason="Invalid or expired JWT token") return client_id = f"sync:{admin.id}:{id(websocket)}" diff --git a/server/api/webssh.py b/server/api/webssh.py index eb7b4eb5..bc2b6566 100644 --- a/server/api/webssh.py +++ b/server/api/webssh.py @@ -63,13 +63,17 @@ async def _verify_webssh_token(token: str): if not admin or not admin.is_active: return None, None - # P1-7: Check updated_at — invalidate token after password change - token_updated = payload.get("updated", 0) - if token_updated and admin.updated_at: - admin_ts = int(admin.updated_at.timestamp()) - # Allow 5-second grace for clock skew / race conditions - if admin_ts > token_updated + 5: + token_tv = payload.get("tv") + if token_tv is not None: + if int(token_tv) != int(admin.token_version or 0): return None, None + else: + # Legacy WebSSH tokens without tv — fall back to updated_at grace + token_updated = payload.get("updated", 0) + if token_updated and admin.updated_at: + admin_ts = int(admin.updated_at.timestamp()) + if admin_ts > token_updated + 5: + return None, None return admin, server_id except Exception: @@ -96,7 +100,7 @@ async def terminal_ws( admin, token_server_id = await _verify_webssh_token(token) if not admin: - await websocket.close(code=4001, reason="Invalid or expired JWT token") + await websocket.close(code=4401, reason="Invalid or expired JWT token") return # Security: WebSSH token must bind to this server (general access_token rejected) diff --git a/server/application/services/auth_service.py b/server/application/services/auth_service.py index 8b53208a..238f634a 100644 --- a/server/application/services/auth_service.py +++ b/server/application/services/auth_service.py @@ -255,14 +255,19 @@ class AuthService: ) return {"success": False, "reason": "invalid_token", "message": "无效的刷新令牌"} - # Rotate: remove old token from Redis, generate new token pair - await self._remove_refresh_token(admin.id, refresh_token) - access_token = self._create_access_token(admin) new_refresh = self._create_refresh_token(admin) - # Store new refresh token in Redis - await self._store_refresh_token(admin.id, new_refresh) + rotated = await self._rotate_refresh_token(admin.id, refresh_token, new_refresh) + if rotated is None: + return { + "success": False, + "reason": "service_unavailable", + "message": "认证服务暂不可用，请稍后重试", + } + if rotated is False: + await self._handle_refresh_token_reuse(admin, ip_address) + return {"success": False, "reason": "invalid_token", "message": "令牌已失效，请重新登录"} return { "success": True, @@ -461,6 +466,7 @@ class AuthService: "iat": now, "exp": now + datetime.timedelta(minutes=JWT_WEBSSH_TOKEN_EXPIRE_MINUTES), "updated": int(admin.updated_at.timestamp()) if admin.updated_at else 0, + "tv": admin.token_version or 0, } return jwt.encode(payload, settings.SECRET_KEY, algorithm="HS256") @@ -618,6 +624,44 @@ class AuthService: membership = await self._refresh_token_membership(admin_id, token) return membership is True + async def _rotate_refresh_token( + self, admin_id: int, old_token: str, new_token: str + ) -> Optional[bool]: + """Atomically swap refresh token hashes in Redis. + + Returns True on success, False if old token absent (reuse/invalid), + None if Redis is unavailable. + """ + try: + from server.infrastructure.redis.client import get_redis + redis = get_redis() + key = f"{REFRESH_TOKEN_REDIS_PREFIX}{admin_id}" + old_hash = self._hash_token(old_token) + new_hash = self._hash_token(new_token) + ttl_seconds = _refresh_token_expire_days() * 86400 + 3600 + script = """ + local removed = redis.call('SREM', KEYS[1], ARGV[1]) + if removed == 0 then + return 0 + end + redis.call('SADD', KEYS[1], ARGV[2]) + local ttl = redis.call('TTL', KEYS[1]) + if ttl < 0 then + redis.call('EXPIRE', KEYS[1], tonumber(ARGV[3])) + end + return 1 + """ + result = await redis.eval(script, 1, key, old_hash, new_hash, str(ttl_seconds)) + if result == 1: + return True + if result == 0: + return False + logger.error("Unexpected Redis rotate result for admin %s: %r", admin_id, result) + return None + except Exception as e: + logger.error(f"Failed to rotate refresh token in Redis for admin {admin_id}: {e}") + return None + async def _remove_refresh_token(self, admin_id: int, token: str): """Remove a single refresh token from Redis Set (single-device logout or rotation)""" try: diff --git a/server/application/services/script_service.py b/server/application/services/script_service.py index 15a8acd5..afda76a1 100644 --- a/server/application/services/script_service.py +++ b/server/application/services/script_service.py @@ -505,6 +505,22 @@ class ScriptService: async def _attach_remote_logs(self, results: Dict[str, Any], tail_lines: int) -> None: """In-place: add remote_log for servers with a nexus-job log path.""" tail_lines = max(10, min(tail_lines, 500)) + server_ids: set[int] = set() + for sid_str, entry in results.items(): + if not isinstance(entry, dict): + continue + if not self._extract_nexus_job_log(entry.get("stdout", "")): + continue + try: + server_ids.add(int(sid_str)) + except ValueError: + continue + + server_map: Dict[int, Server] = {} + for sid in server_ids: + server = await self.server_repo.get_by_id(sid) + if server: + server_map[sid] = server async def _tail_one(sid_str: str, entry: dict) -> None: log_path = self._extract_nexus_job_log(entry.get("stdout", "")) @@ -514,7 +530,7 @@ class ScriptService: server_id = int(sid_str) except ValueError: return - server = await self.server_repo.get_by_id(server_id) + server = server_map.get(server_id) if not server or not server.is_online: entry["remote_log"] = "(服务器离线，无法拉取日志)" return @@ -549,8 +565,14 @@ class ScriptService: results: Dict[str, dict] = {} semaphore = asyncio.Semaphore(settings.SCRIPT_EXEC_CONCURRENCY) + server_map: Dict[int, Server] = {} + for sid in server_ids: + server = await self.server_repo.get_by_id(sid) + if server: + server_map[sid] = server + async def _exec_on_server(server_id: int): - server = await self.server_repo.get_by_id(server_id) + server = server_map.get(server_id) if not server or not server.is_online: results[str(server_id)] = { "status": "skipped", diff --git a/server/application/services/sync_engine_v2.py b/server/application/services/sync_engine_v2.py index 990c3bbd..04e75874 100644 --- a/server/application/services/sync_engine_v2.py +++ b/server/application/services/sync_engine_v2.py @@ -20,12 +20,18 @@ from server.infrastructure.database.sync_log_repo import SyncLogRepositoryImpl from server.infrastructure.database.audit_log_repo import AuditLogRepositoryImpl from server.infrastructure.database.push_schedule_repo import PushRetryJobRepositoryImpl from server.infrastructure.database.crypto import decrypt_value -from server.utils.posix_paths import UPLOAD_STAGING_PREFIX, normalize_remote_abs_path, to_posix +from server.utils.posix_paths import ( + PosixPathError, + UPLOAD_STAGING_PREFIX, + normalize_remote_abs_path, + resolve_nexus_push_source_path, + to_posix, +) def _nexus_source_path(path: str) -> str: - """Normalize push source path string (Nexus host, Linux deployment).""" - return to_posix(path.strip()) + """Normalize and validate push source path on the Nexus host.""" + return resolve_nexus_push_source_path(path) logger = logging.getLogger("nexus.sync_v2") @@ -63,10 +69,11 @@ class SyncEngineV2: batch_id: Optional[str] = None, ) -> dict: """Push files from Nexus to target servers via rsync over SSH""" - source_path = _nexus_source_path(source_path) - if not os.path.isdir(source_path): + try: + source_path = _nexus_source_path(source_path) + except PosixPathError as exc: return {"total": 0, "completed": 0, "failed": 0, "results": {}, - "error": f"源路径不存在: {source_path}"} + "error": str(exc)} batch_id = (batch_id or uuid.uuid4().hex[:12])[:12] concurrency = min(concurrency, MAX_CONCURRENT) @@ -79,6 +86,7 @@ class SyncEngineV2: failed = 0 cancelled = 0 _counter_lock = asyncio.Lock() + _db_lock = asyncio.Lock() results = {} async def _sync_one(server: Server): @@ -102,7 +110,8 @@ class SyncEngineV2: finished_at=datetime.now(timezone.utc), error_message="推送已取消", ) - sync_log = await self.sync_log_repo.create(sync_log) + async with _db_lock: + sync_log = await self.sync_log_repo.create(sync_log) async with _counter_lock: cancelled += 1 @@ -137,7 +146,8 @@ class SyncEngineV2: sync_mode=sync_mode, started_at=datetime.now(timezone.utc), ) - sync_log = await self.sync_log_repo.create(sync_log) + async with _db_lock: + sync_log = await self.sync_log_repo.create(sync_log) try: from server.api.websocket import broadcast_sync_progress @@ -175,42 +185,43 @@ class SyncEngineV2: sync_log.diff_summary = (result.get("stdout") or "")[:2000] else: sync_log.diff_summary = None - sync_log = await self.sync_log_repo.update(sync_log) - # Auto-create retry job for failed pushes (dedup: one pending per server+path) retry_job_id = None - if sync_log.status == "failed": - try: - from server.domain.models import PushRetryJob - from sqlalchemy import select as _select - existing = await self.retry_repo.session.execute( - _select(PushRetryJob).where( - PushRetryJob.server_id == server.id, - PushRetryJob.source_path == source_path, - PushRetryJob.target_path == dest, - PushRetryJob.status == "pending", + async with _db_lock: + sync_log = await self.sync_log_repo.update(sync_log) + + if sync_log.status == "failed": + try: + from server.domain.models import PushRetryJob + from sqlalchemy import select as _select + existing = await self.retry_repo.session.execute( + _select(PushRetryJob).where( + PushRetryJob.server_id == server.id, + PushRetryJob.source_path == source_path, + PushRetryJob.target_path == dest, + PushRetryJob.status == "pending", + ) ) - ) - if existing.scalars().first() is None: - retry_job = PushRetryJob( - server_id=server.id, - server_name=server.name, - operator=operator, - source_path=source_path, - target_path=dest, - status="pending", - retry_count=0, - max_retries=3, - next_retry_at=datetime.now(timezone.utc), - last_error=sync_log.error_message[:500] if sync_log.error_message else None, - ) - retry_job = await self.retry_repo.create(retry_job) - retry_job_id = retry_job.id - logger.info(f"Auto-created retry job #{retry_job.id} for server {server.id}") - else: - logger.debug(f"Retry job already pending for server {server.id}, skipping duplicate") - except Exception as e: - logger.warning(f"Failed to create retry job for server {server.id}: {e}") + if existing.scalars().first() is None: + retry_job = PushRetryJob( + server_id=server.id, + server_name=server.name, + operator=operator, + source_path=source_path, + target_path=dest, + status="pending", + retry_count=0, + max_retries=3, + next_retry_at=datetime.now(timezone.utc), + last_error=sync_log.error_message[:500] if sync_log.error_message else None, + ) + retry_job = await self.retry_repo.create(retry_job) + retry_job_id = retry_job.id + logger.info(f"Auto-created retry job #{retry_job.id} for server {server.id}") + else: + logger.debug(f"Retry job already pending for server {server.id}, skipping duplicate") + except Exception as e: + logger.warning(f"Failed to create retry job for server {server.id}: {e}") async with _counter_lock: if sync_log.status == "success": @@ -313,9 +324,10 @@ class SyncEngineV2: if not server: return {"error": "Server not found", "exit_code": -1} - source_path = _nexus_source_path(source_path) - if not os.path.isdir(source_path): - return {"error": f"源路径不存在: {source_path}", "exit_code": -1} + try: + source_path = _nexus_source_path(source_path) + except PosixPathError as exc: + return {"error": str(exc), "exit_code": -1} dest = target_path or server.target_path or "/tmp/sync" result = await _rsync_push(server, source_path, dest, sync_mode, dry_run=True, verbose=verbose) diff --git a/server/background/retry_runner.py b/server/background/retry_runner.py index f5bf4d1c..37eb15b0 100644 --- a/server/background/retry_runner.py +++ b/server/background/retry_runner.py @@ -37,6 +37,7 @@ def apply_push_retry_outcome( if job.retry_count >= job.max_retries: job.status = "failed" else: + job.status = "pending" job.next_retry_at = now + timedelta(seconds=_backoff_seconds(job.retry_count)) return @@ -54,10 +55,33 @@ def apply_push_retry_outcome( job.last_error = f"Retry exhausted after {job.retry_count} attempts" return + job.status = "pending" job.next_retry_at = now + timedelta(seconds=_backoff_seconds(job.retry_count)) job.last_error = f"Retry {job.retry_count} failed" +async def _claim_next_retry_job(session, now: datetime) -> PushRetryJob | None: + """Claim one due retry job with row lock (SKIP LOCKED).""" + from sqlalchemy import select + + result = await session.execute( + select(PushRetryJob) + .where( + PushRetryJob.status == "pending", + PushRetryJob.retry_count < PushRetryJob.max_retries, + PushRetryJob.next_retry_at <= now, + ) + .limit(1) + .with_for_update(skip_locked=True) + ) + job = result.scalar_one_or_none() + if not job: + return None + job.status = "running" + await session.commit() + return job + + async def retry_runner_loop(): """Background task: every 5min, retry pending retry jobs that are due.""" logger.info("Retry runner loop started (interval: 5min)") @@ -66,23 +90,15 @@ async def retry_runner_loop(): try: async with AsyncSessionLocal() as session: - from sqlalchemy import select now = datetime.now(timezone.utc) - - result = await session.execute( - select(PushRetryJob).where( - PushRetryJob.status == "pending", - PushRetryJob.retry_count < PushRetryJob.max_retries, - PushRetryJob.next_retry_at <= now, - ) - ) - jobs = result.scalars().all() - - if not jobs: - continue - retried = 0 - for job in jobs: + + while True: + job = await _claim_next_retry_job(session, now) + if not job: + break + + job_id = job.id try: from server.application.services.sync_engine_v2 import SyncEngineV2 from server.infrastructure.database.server_repo import ServerRepositoryImpl @@ -110,16 +126,16 @@ async def retry_runner_loop(): if job.status == "completed": logger.info( - f"Retry job {job.id}: success after {job.retry_count} attempts" + f"Retry job {job_id}: success after {job.retry_count} attempts" ) elif job.status == "failed": logger.warning( - f"Retry job {job.id}: failed — {job.last_error}" + f"Retry job {job_id}: failed — {job.last_error}" ) else: backoff = int((job.next_retry_at - now).total_seconds()) logger.warning( - f"Retry job {job.id}: attempt {job.retry_count} pending, " + f"Retry job {job_id}: attempt {job.retry_count} pending, " f"next retry in {backoff}s — {job.last_error}" ) @@ -127,15 +143,20 @@ async def retry_runner_loop(): retried += 1 except Exception as e: - logger.error(f"Retry job {job.id} execution error: {e}") + logger.error(f"Retry job {job_id} execution error: {e}") + await session.rollback() + job = await session.get(PushRetryJob, job_id) + if not job: + continue job.retry_count += 1 job.last_error = str(e)[:500] if job.retry_count >= job.max_retries: job.status = "failed" logger.warning( - f"Retry job {job.id}: max retries reached (exception), marking failed" + f"Retry job {job_id}: max retries reached (exception), marking failed" ) else: + job.status = "pending" job.next_retry_at = now + timedelta( seconds=_backoff_seconds(job.retry_count) ) diff --git a/server/background/schedule_runner.py b/server/background/schedule_runner.py index 34256db6..379cb2e7 100644 --- a/server/background/schedule_runner.py +++ b/server/background/schedule_runner.py @@ -73,6 +73,36 @@ def _field_match(expr: str, value: int) -> bool: return False +def _schedule_is_due(schedule: PushSchedule, now: datetime) -> bool: + """Return True if *schedule* should fire at *now* (cron or once).""" + run_mode = getattr(schedule, "run_mode", None) or "cron" + + if run_mode == "once": + fire_at = getattr(schedule, "fire_at", None) + if not fire_at: + return False + now_naive = now.replace(tzinfo=None) + fire_naive = fire_at.replace(tzinfo=None) if fire_at.tzinfo else fire_at + if fire_naive > now_naive: + return False + if schedule.last_run_at: + return False + return True + + cron_expr = getattr(schedule, "cron_expr", None) + if not cron_expr or not _cron_match(cron_expr, now): + return False + if schedule.last_run_at: + last_run = schedule.last_run_at + now_naive = now.replace(tzinfo=None) + if last_run.tzinfo is not None: + last_run = last_run.replace(tzinfo=None) + seconds_since = (now_naive - last_run).total_seconds() + if seconds_since < SCHEDULE_CHECK_INTERVAL: + return False + return True + + async def schedule_runner_loop(): """Background task: check schedules every 60s, trigger due cron jobs.""" logger.info("Schedule runner loop started (interval: 60s)") @@ -82,67 +112,58 @@ async def schedule_runner_loop(): try: async with AsyncSessionLocal() as session: from sqlalchemy import select - result = await session.execute( - select(PushSchedule).where(PushSchedule.enabled == True) + id_result = await session.execute( + select(PushSchedule.id).where(PushSchedule.enabled == True) ) - schedules = result.scalars().all() + schedule_ids = list(id_result.scalars().all()) now = datetime.now(timezone.utc) triggered = 0 - for schedule in schedules: - run_mode = getattr(schedule, 'run_mode', None) or 'cron' - + for sched_id in schedule_ids: try: - # ── One-time schedule: check fire_at ── - if run_mode == 'once': - fire_at = getattr(schedule, 'fire_at', None) - if not fire_at: - continue - now_naive = now.replace(tzinfo=None) - fire_naive = fire_at.replace(tzinfo=None) if fire_at.tzinfo else fire_at - if fire_naive > now_naive: - continue # not time yet - if schedule.last_run_at: - continue # already ran - else: - # ── Cron schedule: check cron expression ── - cron_expr = getattr(schedule, 'cron_expr', None) - if not cron_expr or not _cron_match(cron_expr, now): - continue - if schedule.last_run_at: - last_run = schedule.last_run_at - now_naive = now.replace(tzinfo=None) - if last_run.tzinfo is not None: - last_run = last_run.replace(tzinfo=None) - seconds_since = (now_naive - last_run).total_seconds() - if seconds_since < SCHEDULE_CHECK_INTERVAL: - continue + lock_result = await session.execute( + select(PushSchedule) + .where( + PushSchedule.id == sched_id, + PushSchedule.enabled == True, + ) + .with_for_update(skip_locked=True) + ) + schedule = lock_result.scalar_one_or_none() + if not schedule: + continue + + if not _schedule_is_due(schedule, now): + await session.rollback() + continue + + run_mode = getattr(schedule, "run_mode", None) or "cron" # Parse server_ids from JSON try: server_ids = json.loads(schedule.server_ids) except (json.JSONDecodeError, TypeError): logger.error(f"Schedule {schedule.id}: invalid server_ids JSON") + await session.rollback() continue if not server_ids: + await session.rollback() continue # Mark as running BEFORE execution to prevent double-trigger - # (if execution takes longer than SCHEDULE_CHECK_INTERVAL). - # On failure we reset last_run_at so the schedule can retry. prev_last_run_at = schedule.last_run_at schedule.last_run_at = now - if run_mode == 'once': + if run_mode == "once": schedule.enabled = False await session.commit() - schedule_type = getattr(schedule, 'schedule_type', 'push') or 'push' + schedule_type = getattr(schedule, "schedule_type", "push") or "push" execution_ok = True try: - if schedule_type == 'script': + if schedule_type == "script": # ── Script execution schedule ── from server.application.services.script_service import ScriptService from server.infrastructure.database.script_repo import ( @@ -212,22 +233,27 @@ async def schedule_runner_loop(): f"Schedule '{schedule.name}' (push) triggered: " f"{result['completed']}/{result['total']} succeeded" ) + if result.get("error"): + execution_ok = False + elif result.get("total", 0) > 0 and result.get("completed", 0) == 0: + execution_ok = False except Exception as exec_err: logger.error(f"Schedule {schedule.id} execution failed: {exec_err}") execution_ok = False if not execution_ok: - rollback_schedule_after_failed_run( - schedule, prev_last_run_at, run_mode - ) - await session.commit() - else: + fresh = await session.get(PushSchedule, sched_id) + if fresh: + rollback_schedule_after_failed_run( + fresh, prev_last_run_at, run_mode + ) await session.commit() triggered += 1 except Exception as e: - logger.error(f"Schedule {schedule.id} setup/parse failed: {e}") + logger.error(f"Schedule {sched_id} setup/parse failed: {e}") + await session.rollback() if triggered: logger.info(f"Schedule runner: {triggered} schedules triggered") diff --git a/server/main.py b/server/main.py index 005d7b5a..87093b34 100644 --- a/server/main.py +++ b/server/main.py @@ -19,7 +19,9 @@ D7: DB session leak fix — middleware auto-manages session lifecycle per reques """ import asyncio +import fcntl import logging +import os from contextlib import asynccontextmanager from pathlib import Path @@ -86,6 +88,20 @@ _background_tasks: list[asyncio.Task] = [] # Redis-based primary worker lock (prevents duplicate background tasks with multi-worker uvicorn) _PRIMARY_LOCK_KEY = "nexus:primary_worker" _PRIMARY_LOCK_TTL = 30 # seconds — must renew periodically +_PRIMARY_FILE_LOCK_FD: int | None = None + + +def _try_acquire_file_primary_lock() -> bool: + """Fallback when Redis is unavailable: one process per host via flock.""" + global _PRIMARY_FILE_LOCK_FD + lock_path = os.environ.get("NEXUS_PRIMARY_LOCK_FILE", "/tmp/nexus-primary.lock") + try: + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600) + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + _PRIMARY_FILE_LOCK_FD = fd + return True + except (OSError, BlockingIOError): + return False async def _acquire_primary_lock() -> bool: @@ -109,9 +125,11 @@ async def _acquire_primary_lock() -> bool: return True return False except Exception as e: - # If Redis fails, assume primary (single-worker mode) - logger.warning(f"Primary lock check failed, assuming primary: {e}") - return True + logger.warning("Primary lock check failed, trying file lock fallback: %s", e) + acquired = _try_acquire_file_primary_lock() + if acquired: + logger.info("Primary worker lock acquired via file fallback") + return acquired async def _renew_primary_lock(): diff --git a/server/utils/posix_paths.py b/server/utils/posix_paths.py index b886b042..673af202 100644 --- a/server/utils/posix_paths.py +++ b/server/utils/posix_paths.py @@ -15,6 +15,19 @@ import posixpath UPLOAD_STAGING_PREFIX = "/tmp/nexus_upload_" +FORBIDDEN_NEXUS_SOURCE_PREFIXES = ( + "/etc", + "/root", + "/home", + "/var", + "/boot", + "/dev", + "/proc", + "/sys", + "/run", + "/srv", +) + class PosixPathError(ValueError): """Invalid or unsafe POSIX path.""" @@ -87,6 +100,20 @@ def resolve_nexus_host_path(path: str) -> str: return os.path.realpath(posix) +def resolve_nexus_push_source_path(path: str) -> str: + """Resolve and validate a Nexus host push source directory.""" + stripped = path.strip() + if ".." in to_posix(stripped).split("/"): + raise PosixPathError("路径不允许包含 '..'") + real_path = resolve_nexus_host_path(stripped) + for prefix in FORBIDDEN_NEXUS_SOURCE_PREFIXES: + if real_path == prefix or real_path.startswith(prefix + "/"): + raise PosixPathError(f"不允许使用系统路径: {prefix}") + if not os.path.isdir(real_path): + raise PosixPathError(f"源路径不存在: {real_path}") + return real_path + + def ensure_under_nexus_upload(path: str) -> str: """Resolve *path* and ensure it stays under ``/tmp/nexus_upload_*``.""" real = resolve_nexus_host_path(path) diff --git a/tests/test_retry_runner_outcome.py b/tests/test_retry_runner_outcome.py index 93f17c4b..1bf485e0 100644 --- a/tests/test_retry_runner_outcome.py +++ b/tests/test_retry_runner_outcome.py @@ -56,3 +56,15 @@ def test_exhausted_retries_marks_failed(): now=datetime.now(timezone.utc), ) assert job.status == "failed" + + +def test_running_job_returns_to_pending_after_retryable_failure(): + job = _job(retry_count=1, status="running") + now = datetime.now(timezone.utc) + apply_push_retry_outcome( + job, + {"total": 1, "completed": 0, "failed": 1, "cancelled": 0}, + now=now, + ) + assert job.status == "pending" + assert job.next_retry_at > now diff --git a/tests/test_security_unit.py b/tests/test_security_unit.py index 634f22fc..8b1d092d 100644 --- a/tests/test_security_unit.py +++ b/tests/test_security_unit.py @@ -227,3 +227,33 @@ def test_install_reject_when_locked(monkeypatch, tmp_path): with pytest.raises(HTTPException) as exc: install_api._reject_if_locked() assert exc.value.status_code == 403 + + +def test_install_verify_token_rejects_missing(monkeypatch, tmp_path): + from fastapi import HTTPException + from server.api import install as install_api + + state_file = tmp_path / ".install_state.json" + state_file.write_text('{"install_token":"abc123"}', encoding="utf-8") + monkeypatch.setattr(install_api, "STATE_FILE", state_file) + + with pytest.raises(HTTPException) as exc: + install_api._verify_install_token("") + assert exc.value.status_code == 403 + + with pytest.raises(HTTPException) as exc: + install_api._verify_install_token("wrong") + assert exc.value.status_code == 403 + + install_api._verify_install_token("abc123") + + +def test_resolve_nexus_push_source_path_rejects_system_prefix(tmp_path, monkeypatch): + from server.utils.posix_paths import PosixPathError, resolve_nexus_push_source_path + + allowed = tmp_path / "push_src" + allowed.mkdir() + assert resolve_nexus_push_source_path(str(allowed)) == str(allowed.resolve()) + + with pytest.raises(PosixPathError, match="系统路径"): + resolve_nexus_push_source_path("/etc") diff --git a/web/app/install.html b/web/app/install.html index 1750d70a..56fa0947 100644 --- a/web/app/install.html +++ b/web/app/install.html @@ -566,6 +566,7 @@ function installWizard() { envChecks: [], envAllPass: false, initResult: null, + installToken: '', installDir: '/opt/nexus', btPanel: false, @@ -659,6 +660,7 @@ function installWizard() { if (r.ok && d.success) { this.success = '数据库初始化成功！已创建 ' + (d.tables_created || 14) + ' 张表。'; this.initResult = d; + this.installToken = d.install_token || ''; this.btPanel = !!d.bt_panel; this.step = 4; } else { @@ -675,6 +677,7 @@ function installWizard() { this.error = ''; try { const body = { + install_token: this.installToken, ...this.adminForm, db_host: this.form.db_host, db_port: this.form.db_port,