Skip to content

Q
qarm_setup_scripts
Commit 5b148717 by jaysurplus@126.com
Options

docs(iteration): 归档项目迭代 V0.4.5 - 批量注册与持续 TUI

parent 85b2d526
Showing with 1207 additions and 49 deletions
.gitignore
......@@ -17,3 +17,5 @@ Thumbs.db
*.swp
__pycache__/
*.pyc
*.local.json
*.local.json.lock
README.md
......@@ -23,7 +23,8 @@ QArm 必要配置与环境初始化脚本仓库。
| 脚本 | 用途 | 文档 |
| --- | --- | --- |
| `scripts/tailscale/tailscale-route-manager.sh`、`tailscale-oauth-helper.py` | 在目标 Linux 设备安装和注册 Tailscale，选择设备标签，并管理 `192.168.11.0/24`、`192.168.22.0/24` 子网路由 | `docs/tailscale/README.md` |
| `scripts/tailscale/tailscale-route-manager.sh`、`tailscale-oauth-helper.py` | 在目标 Linux 设备安装和注册 Tailscale，管理 SN 网络名称、标签及互斥的 4via6/旧 IPv4 路由模式 | `docs/tailscale/README.md` |
| `scripts/tailscale/tailscale-batch-register.py` | 从控制电脑通过 SSH 批量注册，核对 SN、生成网络名称并分配 Site ID | [批量注册](docs/tailscale/batch-registration.md) |
## Agent 初始化
......
docs/tailscale/README.md
......@@ -12,8 +12,8 @@
- 检测、下载并安装 Tailscale。
- 检测 `tailscaled` 和登录状态，必要时运行 `tailscale up`。
- 提供独立的 Tailscale 登录和退出登录操作；登录时可选择网页认证或 OAuth 自动注册，退出登录前需要确认。
- 使用 OAuth Client 在线读取可分配标签，在 TUI 中输入设备名称并多选标签，自动注册新设备。
- 注册名称默认从 `/opt/armaster/data/hw.json` 读取 `sn` 并生成 `<sn>-` 前缀，文件不可用时回退到系统 hostname。
- 使用 OAuth Client 在线读取可分配标签，在 TUI 中输入设备用途并多选标签，自动注册新设备。
- 注册和改名都必须从 `/opt/armaster/data/hw.json` 读取 `sn`，并根据用户输入的用途生成 `<硬件SN>-<用途>`；SN 缺失或无效时拒绝操作。
- 每台设备动态创建 10 分钟有效、单次使用、非临时且预批准的 Auth Key。
- 启用 IPv4 forwarding，并在系统支持时启用 IPv6 forwarding。
- 支持“仅 4via6（推荐）”和“仅旧 IPv4”两种互斥模式；切换前必须先禁用当前路由。
......@@ -26,6 +26,7 @@
- 优先使用支持逐段着色的 `dialog`，其次使用 `whiptail`，否则使用普通终端菜单。
- 对安装、运行、连接和 Route 状态使用统一的状态颜色。
- 状态页显示当前设备在 Tailscale 网络中的名称。
- 支持在 TUI 或命令行中修改 Tailscale 网络名称；MagicDNS 名称随之更新，但不会修改 Linux 系统 hostname。
- 主菜单提供独立的“网络检测”，完整显示 `tailscale netcheck` 的结果并返回主页面。
## TUI 导航规则
......@@ -35,7 +36,8 @@
- 在分支流程中选择“否”、取消、关闭提示框或操作失败时，也会返回主页面。
- 在 `whiptail` 或 `dialog` 主菜单按 Esc 不会退出脚本，而是重新显示主页面。
- 在“是否现在运行 `tailscale up`”中选择 No，会取消本次启用并直接返回主页面。
- `dialog`/`whiptail` 下选择菜单操作后会立即显示中文处理中界面，网络检测、状态读取和 OAuth 请求期间不会短暂露出底层终端。
- 网络检测、完整状态读取和主菜单刷新使用持续运行的 `dialog`/`whiptail` gauge 窗口覆盖等待过程，命令结束后才关闭窗口并显示结果。单次 `infobox` 会立即退出，在 `whiptail` 下不能用来覆盖耗时命令。
- 主菜单会话在有 `tic` 和 `tput` 时使用临时 terminfo，统一管理进入/离开 TUI 屏幕。菜单、进度和结果窗口之间不再反复恢复 shell；从主页面退出时恢复终端并删除临时配置。不会修改系统 terminfo；缺少这些工具时退回单窗口模式，仍可能看到短暂切换。
- 网页登录和首次安装器执行需要展示认证 URL 或安装日志，因此这两类明确的交互仍会进入终端；结束后返回主页面。
- `install`、`enable`、`disable` 等命令行直接调用模式不进入 TUI 循环，仍通过退出码向调用方报告结果。
- 状态页显示脚本版本，可用于确认目标设备是否仍在运行旧副本。
......@@ -91,6 +93,7 @@ sudo tailscale-route-manager
sudo tailscale-route-manager install
sudo tailscale-route-manager route-mode 4via6
sudo tailscale-route-manager site-id 101
sudo tailscale-route-manager hostname qarm
sudo tailscale-route-manager login
sudo tailscale-route-manager logout
sudo --preserve-env=TS_OAUTH_CLIENT_ID,TS_OAUTH_CLIENT_SECRET \
......@@ -103,6 +106,8 @@ sudo tailscale-route-manager disable
## OAuth 自动注册
通过局域网 SSH 无人值守批量注册、自动分配 Site ID，参见 [批量注册说明](batch-registration.md)。控制端负责创建单次 Auth Key，设备只接收临时密钥。
OAuth 自动注册适用于批量部署新的、尚未连接到 Tailscale 的 QArm 设备。管理器不会修改 Tailnet Policy；管理员必须提前完成以下配置：
1. 在 Tailnet Policy 中定义 Provisioner 标签和设备标签的直接所有权：
......@@ -143,7 +148,7 @@ OAuth 自动注册适用于批量部署新的、尚未连接到 Tailscale 的 QA
Provisioner 和设备标签建议全部使用小写，例如 `tag:qarm`。Auth Key 标签与 `tailscale up --advertise-tags` 必须完全一致，避免混合大小写被控制端拒绝。
设备名称输入框会优先读取 `/opt/armaster/data/hw.json` 的 `sn`，例如默认显示 `CA2EUP-20261020-001-`，用户在末尾补充名称即可。脚本只读该文件；文件缺失、JSON 无效、SN 不符合 DNS 标签字符规则或没有后缀空间时，默认值回退到系统 hostname。
设备用途输入框会先读取 `/opt/armaster/data/hw.json` 的 `sn`，例如显示前缀 `CA2EUP-20261020-001-`。用户只输入 `qarm` 等用途，脚本生成 `CA2EUP-20261020-001-qarm`。脚本只读该文件；文件缺失、JSON 无效、SN 不符合 DNS 标签字符规则或完整名称超过 63 个字符时，注册会停止，不使用系统 hostname 或临时局域网 IP 作为替代身份。
注册流程不会把 OAuth Secret、Access Token 或 Auth Key 写入仓库和持久状态。短期 Auth Key 会先写入权限为 `0600` 的临时文件，供 `tailscale up` 读取，并在命令结束或脚本收到退出信号时删除。Auth Key 已创建但注册失败时，未使用的 Key 最多在 10 分钟后过期。
......@@ -159,6 +164,18 @@ logout 后可以使用任一种方式重新认证。OAuth 方式按新设备流
登录和退出登录只管理 Tailscale 身份状态。退出登录不会自动移除 advertised routes，也不会恢复 IPv4/IPv6 forwarding；如需清理本脚本管理的子网路由和转发状态，应先选择“禁用子网路由”。
## 修改 Tailscale 网络名称
主菜单选择“修改 Tailscale 网络名称”，输入框会显示从当前 SN 前缀名称中提取的用途；旧名称不符合 SN 前缀时留空。也可以直接传入新用途：
```bash
sudo tailscale-route-manager hostname qarm
```
命令参数是“用途”而不是完整名称。脚本读取本机硬件 SN，组合为 `<硬件SN>-<用途>`；用途只能包含字母、数字和连字符，完整名称不得超过 63 个字符。它使用 `tailscale set --hostname=...` 修改 Tailscale machine name；MagicDNS 名称会同步变化，Linux 系统 `/etc/hostname` 不会被修改。设备必须已经登录 Tailscale，脚本会在写入后重新读取状态确认名称。
局域网 IP（例如 `192.168.81.118`）只用于设备首次接入和 SSH 引导，不是设备长期身份，也不会进入 Tailscale 网络名称。设备部署到不同地点后，应通过 `<硬件SN>-<用途>` 的 MagicDNS 名称或 Tailscale IP 继续管理。
## 安装安全边界
脚本不会使用 `curl | sh`。它会：
......
docs/tailscale/batch-registration.md 0 → 100644
# 局域网 SSH 批量注册
`scripts/tailscale/tailscale-batch-register.py` 在控制电脑上读取设备清单，自动核对硬件 SN、创建一次性 Auth Key、设置 `<硬件SN>-<用途>` 名称、分配 Site ID，并可选启用 4via6 路由。默认只读预检，只有 `--apply` 才修改设备。
## 前置条件
- 控制电脑：Windows 或 Linux、Python 3、OpenSSH `ssh`，同时保留该目录的 `tailscale-batch-device.py`、`tailscale-oauth-helper.py` 和 `tailscale-route-manager.sh`。
- 目标设备：Linux、Python 3、Bash、`sysctl` 等原管理器依赖、已安装并运行的 Tailscale，且 `tailscale debug via` 可用。工具不自动安装软件或启动服务。
- 已配置 SSH 公钥登录；管理员应先核实服务器 host key 并存入 known_hosts。工具使用 `BatchMode=yes` 与 `StrictHostKeyChecking=yes`，不会接受未知或已改变的 host key。
- SSH 用户可以通过 `sudo -n python3` 非交互执行设备工作器，属于 root 权限；工具不创建 sudo 规则，也不接收密码。不能为方便而给可被普通用户修改的脚本配置免密 root 执行。
- 新设备注册需要控制端提供 `TS_OAUTH_CLIENT_ID`、`TS_OAUTH_CLIENT_SECRET`，权限/tagOwners 与现有 OAuth 注册文档一致。已登录设备不需要新 Auth Key，保留原身份和标签；仅更新网络名称与本工具管理的配置。
## 1. 准备清单与持久注册表
复制 `scripts/tailscale/batch-inventory.example.json` 为本地清单，例如 `devices.local.json`。示例使用保留地址，必须替换为实际连接地址、SSH 用户、真实硬件 SN、用途和允许的标签。
### devices.local.json：本次要处理的设备
文件顶层是 JSON 数组，每个对象代表一台设备。以下内容可作为完整文件模板：
```json
[
{
"host": "192.0.2.10",
"user": "operator",
"sn": "QARM-EXAMPLE-001",
"purpose": "coffee",
"tags": ["tag:qarm"]
},
{
"host": "192.0.2.11",
"user": "operator",
"sn": "QARM-EXAMPLE-002",
"purpose": "maintenance",
"site_id": 101,
"tags": ["tag:qarm", "tag:europe"]
}
]
```
| 字段 | 必填 | 内容 |
| --- | --- | --- |
| `host` | 是 | 当前可达的局域网 IP、Tailscale IP 或 DNS 名称；搬迁后可以修改 |
| `user` | 是 | 设备上的 SSH 用户名，需具备非交互 sudo 权限 |
| `sn` | 是 | 本机 `hw.json` 中的真实 SN，大小写必须完全一致 |
| `purpose` | 是 | 用途后缀，如 `coffee`；只包含字母、数字、连字符，首尾为字母或数字 |
| `tags` | 否 | 注册新设备时的标签数组，默认 `["tag:qarm"]`；需小写且已获 OAuth 授权；已登录设备保留原标签 |
| `site_id` | 否 | 整数 `1-65535`；省略后复用注册表已有编号或自动分配新编号 |
| `port` | 否 | SSH 端口，整数，默认 `22` |
| `hw_config` | 否 | 设备端硬件 JSON 绝对路径，默认 `/opt/armaster/data/hw.json` |
第一台的预期名称是 `qarm-example-001-coffee`，第二台是 `qarm-example-002-maintenance`。SN 与用途组合最长 63 字符；临时 IP 不参与名称生成。JSON 不支持注释，端口和 Site ID 应写成数字，不能加引号。
### sites.local.json：长期保存的编号注册表
文件顶层是 JSON 对象，键是硬件 SN，值是整数 Site ID。它保存所有已分配设备，不只保存当前批次：
```json
{
"QARM-EXAMPLE-001": 1,
"QARM-EXAMPLE-002": 101
}
```
可以从不存在的 `sites.local.json` 开始，首次 apply 会创建它。已有设备的 Site ID 必须先导入，不能让它们的编号被新设备占用。所有批次应共享并备份同一注册表；工具只检查该表与本批次，不能发现 Tailnet 中未录入的分配。不要删除失败设备的编号；重试会复用它。
如果全部是尚未分配过编号的新设备，也可以手动创建以下完整内容：
```json
{}
```
空注册表与上述清单一起执行 apply 后，第一台自动获得 `1`，第二台使用显式指定的 `101`。如果某台已有 Site ID `1`，应将**该设备真实 SN** 对应到 `1` 后再运行；不能为了通过预检把已有编号改掉。
两个文件都放在运行批量工具的控制电脑上，不需要逐台上传，也不应包含 Auth Key、OAuth Secret 或 SSH 密码。`--inventory` 与 `--registry` 的相对路径按当前工作目录解释。
在已部署设备的脚本目录中，例如先执行 `cd ~/qarm_setup_scripts/scripts/tailscale`，创建好这两个文件后使用：
```bash
python3 ./tailscale-batch-register.py --inventory ./devices.local.json --registry ./sites.local.json --dry-run
```
设备搬迁后只需更新 `host`；SN 和 Site ID 保持不变。后续 host 可以使用可达的 Tailscale IP 或 MagicDNS 名称。改名可能令旧 MagicDNS host 失效，首次改名优先使用当前可达 IP。
## 2. 只读预检
在仓库根目录运行（Windows 可将 `python` 换成已安装的解释器路径）：
```bash
python scripts/tailscale/tailscale-batch-register.py \
--inventory devices.local.json --registry sites.local.json --dry-run
```
工具会通过 SSH 核对 SN、登录状态、Tailscale 能力和已有配置，展示预计网络名称和 Site ID。预检不会创建 Auth Key、写入注册表或修改设备配置。`--enable-routes` 可与预检一起使用以表达下一步意图。
如果设备已配置 legacy 或不同 Site ID，预检拒绝继续；应先核对原配置，再人工禁用/迁移。不会自动中断已有路由。任何设备预检失败，整个批次均不执行 apply。
## 3. 执行注册
在控制电脑环境中注入 OAuth 凭证；不要写入清单、命令参数或仓库。PowerShell 示例：
```powershell
$env:TS_OAUTH_CLIENT_ID = '<运行时 Client ID>'
$env:TS_OAUTH_CLIENT_SECRET = '<运行时 Secret>'
python scripts/tailscale/tailscale-batch-register.py --inventory devices.local.json --registry sites.local.json --apply --enable-routes --report results.local.json
```
`--apply` 注册/改名并保存 4via6 模式与 Site ID。只有增加 `--enable-routes` 才开启 forwarding 和发布两个维护地址的 4via6 路由；未指定时现有路由保持原样。`--identity-file PATH` 可指定控制端 SSH 私钥路径，不会复制私钥到设备。
控制端串行处理设备。OAuth Secret 和 Access Token 留在控制端；每台未登录设备使用独立的 600 秒单次非临时 Auth Key，通过 SSH stdin 传送。设备密钥临时文件权限为 0600，命令成功或失败都会删除。工作器在内存执行，管理器副本在 root 临时目录执行后清理，不覆盖设备上已有脚本。
## 结果、影响与恢复
- 退出码 `0`：所有设备预检/执行成功；`1`：有设备失败或可能部分完成；`2`：参数、清单、注册表或锁文件问题。
- JSON 报告以 SN 为主键内容，包含每台设备的阶段、结果、网络名称、Tailscale IP、完整 DNS 名称和已发布路由；不包含凭证。未返回原始 stderr，以免外部程序输出密钥。
- apply 前先保存编号。注册失败仍保留编号，避免重试时重新分配。注册表存在 `.lock` 时拒绝第二个写入者；异常终止遗留锁时，先确认没有控制端仍运行，再人工移除该单个锁文件。
- 操作不跨设备回滚。超时或中断时设备可能已注册、改名或完成部分路由操作；不要直接重试 logout，先重新 dry-run，再使用同一注册表重试 apply。已经注册的设备会复用身份。
- 密钥的普通失败清理依赖工作器的 finally；若设备断电或工作器被强制终止，可能遗留 root 所有的 0600 临时文件。该 Key 在 600 秒后过期；应由管理员核对并清理对应临时文件，必要时后台撤销尚未使用的 Key。
- forwarding 的影响和恢复沿用原管理器。需要撤销路由时，在对应设备使用原管理器的 `disable`；恢复旧名称可使用管理器 `hostname <原用途>`。不自动注销设备，也不删除中央编号。
- 首次发布的路由仍需要后台批准或已有 autoApprovers；报告“已发布”不等于后台批准，更不等于维护页面可访问。需要从维护客户端访问并验收。
## 最短验证
```bash
python -m unittest discover -v -s tests/tailscale -p 'test_*.py'
python scripts/tailscale/tailscale-batch-register.py --help
```
真实投用先只选一台可核对 SN 的设备执行 dry-run，然后 apply 并读回名称、Site ID、路由与维护页面，再扩大批次。当前自动化测试使用隔离替身，不注册真实设备。
meta/ARCHITECTURE.md
......@@ -29,7 +29,7 @@
- 启用任一模式时移除另一模式的受管路由，同时保留设备上其他 advertised routes。
- 在启用子网路由时配置 IP forwarding，在禁用时恢复脚本执行前的状态。
- 通过 `whiptail`、`dialog` 或普通终端提供中文单出口 TUI。
- 从 `/opt/armaster/data/hw.json` 只读获取 `sn` 作为设备名默认前缀，收集用户补充名称和标签选择，并调用 helper 完成 OAuth 自动注册；配置缺失或无效时回退到系统 hostname。
- 从 `/opt/armaster/data/hw.json` 只读获取 `sn`，与用户提供的用途组成稳定的 `<硬件SN>-<用途>` Tailscale 名称，并调用 helper 完成 OAuth 自动注册；SN 缺失或无效时拒绝注册和改名。临时局域网 IP 只用于首次 SSH 引导，不作为设备身份。
`tailscale-oauth-helper.py` 的窄接口只提供“列出可分配标签”和“创建一次性 Auth Key”。它负责 OAuth 令牌交换、Tailscale API JSON 契约和错误脱敏；不负责 TUI、系统服务、路由或 forwarding。
......@@ -45,7 +45,7 @@
```text
主页面
-> 安装/检查 Tailscale
-> OAuth 注册（设备名称 + 在线标签多选）或浏览器登录
-> OAuth 注册（硬件 SN + 设备用途 + 在线标签多选）或浏览器登录
-> 选择互斥路由模式
-> 4via6：配置唯一 Site ID 并生成两个 IPv4 /32 对应的 routes
-> 旧 IPv4：使用两个历史 IPv4 /24 routes
......@@ -63,3 +63,13 @@
```
TUI 分支中的取消、Esc 和失败均返回主页面；只有主页面“退出”是正常退出入口。CLI 直接调用模式保留退出码语义，便于自动化调用。
## 局域网批量注册
`tailscale-batch-register.py` 是 Windows/Linux 控制端入口：清单和持久 SN → Site ID 注册表 → 整批 SSH 预检 → 保存编号 → 控制端 OAuth 创建单次 Key → 逐设备 apply → JSON 汇总。注册表使用互斥锁与原子替换；失败保留编号以支持重试。host 是连接参数，SN 是资产标识。
`tailscale-batch-device.py` 通过 SSH 的 `sudo -n python3` 临时执行，只接收 JSON stdin；不安装后台服务。设备端再次核对 SN、模式与 Site ID，注册/改名后复用管理器的路由与 forwarding 实现，并回读结果。执行器和管理器临时副本不作为长期部署文件。只有 `--enable-routes` 才发布路由。
OAuth 是外部 API 依赖，复用已有 helper；SSH 是自有执行器的 transport，隔离测试在该边界替代设备。编号分配为进程内计算，注册表在文件系统边界验证持久化与锁；未添加通用 transport 抽象。删除控制端会把分配、凭证和整批失败处理分散到操作人员；删除设备端会把本机 root 状态链路暴露到 SSH 调用方，因此两端分别保留职责。既有 TUI 和 OAuth helper 接口不迁移，不执行多方案设计探索。
限制：预检不代表跨设备事务；超时可能部分生效。已登录设备保留身份/标签，不强制重认证；已有 legacy 或冲突 Site ID 需人工迁移。未录入中央注册表的 Site ID 分配无法自动发现，首次投用须导入已有编号。
meta/FEATURES.md
......@@ -4,5 +4,6 @@
| --- | --- | --- |
| Agent 治理骨架 | Done | AGENTS、workflow、rule、skill 与多 Agent 适配层已初始化 |
| 脚本安全约束 | Done | 已定义敏感信息、幂等、权限、危险操作与验证边界 |
| Tailscale Route Manager | Done (V0.4.2) | 支持 OAuth 批量注册、默认 QArm 标签、SN 设备名前缀、登录/退出、网络检测，以及互斥的 4via6 与旧 IPv4 路由模式；多设备批量注册与 TUI 视觉效果仍待现场复验 |
| Tailscale Route Manager | Done (V0.4.5) | 注册和改名强制使用硬件 SN 与用途；TUI 通过临时 terminfo 保持同一屏幕，已在 118 设备完成伪终端验证，人工视觉验收待确认 |
| 局域网 SSH 批量注册 | Done (V0.4.5) | 支持 SN 核对、自动分配并持久保存 Site ID、单次密钥传输、整批预检、失败汇总和重试；真实批量入网待清单与试点 |
| 其他 QArm 业务配置脚本 | Pending | 等待后续按真实目标平台和配置需求添加 |
meta/ITERATIONS.md
# 迭代记录
## [V0.4.5] SN 命名、局域网批量注册与持续 TUI (SN Naming, LAN Provisioning and Persistent TUI)
- **功能**: 注册和改名统一使用 `<硬件SN>-<用途>`；新增控制端 SSH 批量注册，按 SN 持久分配 Site ID，支持只读预检、失败重试和 JSON 报告；修复网络检测及窗口间短暂恢复终端的问题。
- **改动文件**:
- `scripts/tailscale/tailscale-route-manager.sh`: 新增网络改名、强制 SN 命名、持续 gauge，以及临时 terminfo 屏幕会话。
- `scripts/tailscale/tailscale-batch-register.py`、`tailscale-batch-device.py`: 分离控制端编号/OAuth 与设备端配置，密钥通过 SSH stdin 传输，复用原路由实现。
- `scripts/tailscale/batch-inventory.example.json`: 提供非敏感清单样例。
- `tests/tailscale/`: 覆盖 SN 命名、批量分配/预检/密钥清理，以及真实 whiptail 完整会话的屏幕切换。
- `docs/tailscale/`、`README.md`: 补充清单和注册表的完整 JSON 格式、使用步骤、恢复及 TUI 前置条件。
- `meta/specs/tailscale-lan-batch-registration.md`、`meta/plans/archive/tailscale-lan-batch-registration_build_plan.md`: 固化范围、接口及已完成实现计划。
- **影响**: 设备搬迁后可保持 SN 和 Site ID，连接地址单独更新。控制端保存 OAuth Secret，未登录设备仅接收短期单次 Key。`0.4.5` 已部署到 118 设备，当前旧 IPv4 路由和转发保持不变；该设备上旧版完整流程 7 次屏幕切换已由新版一次进入/退出替代。Linux 26 项 Python 测试和 Bash 回归通过；真实批量注册、维护页面及用户终端视觉效果仍需现场验收。
## [V0.4.2] 4via6 路由模式与 TUI 完善 (Route Modes and TUI Polish)
- **功能**: 增加网页登录与 OAuth 登录选择、退出登录、默认 `tag:qarm`、Tailscale 网络名称和独立 `netcheck`；将重复 IPv4 维护地址升级为 Site ID 驱动的 4via6 路由，并保留互斥的旧 IPv4 兼容模式；为 TUI 分支增加处理中界面，减少操作间露出底层终端。
......
meta/PROJECT_MANIFEST.json
{
"project_name": "qarm_setup_scripts",
"version": "0.4.2",
"iteration_count": 5,
"last_update": "2026-10-25",
"version": "0.4.5",
"iteration_count": 6,
"last_update": "2026-10-26",
"stack_summary": [
"PowerShell",
"Shell",
......@@ -22,7 +22,10 @@
"scripts/sync-agent-adapters.sh",
"scripts/tailscale/tailscale-route-manager.sh",
"scripts/tailscale/tailscale-oauth-helper.py",
"scripts/tailscale/tailscale-batch-register.py",
"scripts/tailscale/tailscale-batch-device.py",
"docs/tailscale/README.md",
"docs/tailscale/batch-registration.md",
"meta/REGRESSION.md",
"meta/RELEASE_CHECKLIST.md"
],
......@@ -45,6 +48,10 @@
"tailscale_default_qarm_tag_ready": true,
"tailscale_network_diagnostics_ready": true,
"tailscale_route_modes_ready": true,
"tailscale_tui_transition_ready": true
"tailscale_tui_transition_ready": true,
"tailscale_sn_purpose_naming_ready": true,
"tailscale_lan_batch_registration_ready": true,
"tailscale_site_registry_ready": true,
"tailscale_persistent_tui_screen_ready": true
}
}
meta/REGRESSION.md
......@@ -23,8 +23,10 @@
| 目标设备端到端路由 | 从 tailnet 客户端访问两个 QArm 局域网 | Pending |
| Tailscale 状态颜色 | `dialog` 与交互终端显示彩色圆点和中文状态，`whiptail` 仅显示中文状态，重定向或 `NO_COLOR` 时为纯文本 | Ready |
| Tailscale 网络名称 | 已连接时状态页显示 `Self.HostName`；未连接时显示“未连接”，字段缺失时显示“未知” | Ready |
| Tailscale 网络名称修改 | 仅接受合法 DNS 标签名称；使用 `tailscale set --hostname`；相同名称不重复写入；写入后重新读取确认 | Ready |
| Tailscale 网络检测 | 主菜单和 `netcheck` 命令显示完整检测输出；失败信息可见且 TUI 返回主页面 | Ready |
| Tailscale TUI 操作过渡 | `dialog`/`whiptail` 菜单分支在执行前显示处理中界面；窗口间不主动清屏暴露底层终端 | Ready |
| Tailscale TUI 操作过渡 | 网络检测/状态读取/主菜单刷新使用持续 gauge；真实 whiptail 伪终端验证耗时命令结束前不恢复底层终端；错误输出与退出码保留 | Ready |
| Tailscale TUI 屏幕会话 | 具备 tic/tput 时完整菜单→检测→结果→菜单→退出流程仅一次进入/离开 alternate screen；临时 terminfo 清理且不修改宿主终端环境 | Ready |
| Whiptail 边框宽度 | 状态文本不得包含双列 emoji；右侧边框不因状态行发生错位 | Ready |
| Tailscale TUI 单出口 | 分支完成、取消、Esc 或失败后返回主页面；仅主页面“退出”结束 TUI | Ready |
| Tailscale 登录 | 未登录设备可执行 `tailscale login`；已登录设备不重复操作；完成后返回主页面 | Ready |
......@@ -38,5 +40,12 @@
| OAuth Auth Key 描述 | 描述只包含 API 接受的 ASCII 字母数字和连字符，且不超过 50 字符 | Ready |
| OAuth 标签大小写 | Provisioner 与设备标签统一使用小写，Auth Key 标签和 `tailscale up` 请求必须一致 | Ready |
| OAuth 默认 QArm 标签 | `tag:qarm` 可用时自动包含它，用户可增加其他标签；不存在时保持普通多选 | Ready |
| 硬件 SN 名称前缀 | 有效 `hw.json.sn` 生成 `<sn>-` 默认值；缺失、损坏、非法或过长时回退 hostname | Ready |
| 硬件 SN 名称前缀 | 注册和改名均由有效 `hw.json.sn` 与用途生成 `<硬件SN>-<用途>`；SN 缺失、损坏、非法或名称过长时拒绝操作 | Ready |
| OAuth 批量真实注册 | 在至少两台新设备连续运行并核对名称、标签和连接状态 | Pending |
| SSH 批量清单 | SN、host、用途、标签、端口与 Site ID 校验；冲突在 SSH 写入前拒绝 | Ready |
| Site ID 中央分配 | 显式编号优先、自动编号唯一、SN 搬迁复用；锁阻止同时写入，失败保留编号 | Ready |
| SSH 批量只读预检 | dry-run 不写设备/注册表、不创建 Auth Key；任一设备预检失败拒绝整批 apply | Ready |
| SSH 批量密钥隔离 | OAuth Secret 留控制端；单次 Key 仅走 SSH stdin；临时文件成功/失败清理；报告不含 Key | Ready |
| SSH 批量已注册设备 | 复用原身份、不创建 Key；改名与配置后读回；冲突路由模式拒绝自动迁移 | Ready |
| SSH 批量跨平台隔离测试 | Windows/Linux 分别运行 Python 测试；Linux 验证 0600 文件权限 | Ready |
| SSH 批量真实设备试点 | 正确 SN、免交互 SSH/sudo、OAuth 注册、后台路由批准及维护页面验收 | Pending |
meta/plans/archive/tailscale-lan-batch-registration_build_plan.md 0 → 100644
# Plan: 局域网批量注册
- 最终状态：Done（本地实现与隔离验证范围）。
- 归档日期：2026-10-26。
- 最终验证：Linux Python 26/26、Windows Python 24 项通过且 2 项 Linux TUI 测试跳过，Bash 回归、语法及 diff 检查通过；118 设备部署和伪终端验证通过。尚未执行真实批量注册。
- 相关 Spec：`meta/specs/tailscale-lan-batch-registration.md`。
- 相关提交：本轮 `docs(iteration)` V0.4.5 提交；尚无独立 PR。
## 1. 来源 Spec
`meta/specs/tailscale-lan-batch-registration.md`；用户授权本地实现，未提供真实批量目标。
## 2. 执行阶段
一个实现 slice：新增控制端和设备端执行器，复用现有 OAuth helper/路由管理器；增加隔离测试、样例及文档。完成标准为本地行为测试通过、支持只读预检、执行和失败汇总，不包含真实环境验收。
## 3. 执行顺序
接口规格 → 两个执行模块 → 隔离测试 → 文档与验证记录。
## 4. 并行与串行判断
本地顺序开发；批量执行串行，单设备超时后汇总继续，不开启并发注册。
## 5. 风险与暂停条件
真实清单缺失不阻塞实现，但禁止代为选择设备。预检任何失败拒绝整个批次 apply；已开始执行后的失败可能留下部分注册/配置状态，保留编号与报告供重试。
## 6. 当前进度表
| Phase | 状态 | 完成日期 | 下一步 |
| --- | --- | --- | --- |
| 批量注册实现与隔离验证 | Done | 2026-10-26 | 提供清单后单设备试点 |
进度详情：新增控制端、设备工作器、示例清单、文档和回归用例；Windows/Linux Python 单元测试均通过（24 项，其中批量注册 16 项），包含 Linux 临时密钥 0600 与失败清理；原 Bash 功能回归通过。`py_compile`、帮助入口与 `git diff --check` 通过。初次 Windows 权限断言因 NTFS 不提供 POSIX mode 而失败，已改为在 Linux 做该断言，并保留 Windows 清理验证。没有执行真实 SSH/OAuth 注册或改动设备；免密 sudo、真实端到端路由与批量投用待试点。
## 7. 重启或上下文恢复后的接续步骤
读取规格、迭代记录与当前 Git 状态；SN 命名和批量功能已纳入 V0.4.5，真实设备试点应以新的现场状态为准。
## 8. 下一步建议
本地完成后，提供实际清单和已有 Site ID 分配，再执行只读预检与单设备试点。
meta/specs/tailscale-lan-batch-registration.md 0 → 100644
# 局域网批量注册 Spec
## 目标与范围
用户已要求实现局域网 SSH 批量注册。控制端使用 Python 3/OpenSSH，读取 JSON 清单的连接地址、SSH 用户、预期硬件 SN、用途、标签和可选 Site ID。名称为 `<硬件SN>-<用途>`。控制端持久注册表以 SN 为键，自动分配 1-65535 中未占用编号；已有设备编号必须导入，不能宣称检测整个 Tailnet 的 Site ID 冲突。
## 非目标与前置条件
不安装软件、不配置 SSH/sudo、不自动批准后台 routes、不开放 HTTP API、不切换已有路由模式、不替用户执行真实批量注册。目标要求 Linux、Python 3、Bash、已运行的 tailscaled、支持 `debug via` 的 Tailscale，以及通过已信任 host key 的非交互 SSH 和 `sudo -n`。
## 接口契约
控制端 `tailscale-batch-register.py --inventory FILE --registry FILE [--apply] [--report FILE]` 默认只读预检；`--dry-run` 与默认行为相同。`--apply` 才修改设备与注册表。注册表必须持久保留，使用互斥锁避免两个 apply 同时分配；失败保留编号，重试复用。报告不包含任何凭证。
## 状态流与失败边界
完整校验清单/重复 SN、地址和编号 → 读取注册表 → 按 SN 分配编号 → SSH 只读核对每台 SN、登录状态与现有路由配置 → 全部预检通过后保存编号 → 控制端 OAuth 创建短期一次性 Auth Key → SSH stdin 传入 → 设备 root 临时文件注册 → 删除临时密钥 → 复用现有 Bash 路由实现配置 4via6/Site ID/转发 → 回读名称/路由/IP → 汇总。
OAuth Secret 只在控制端；Auth Key 不进入 SSH 参数、报告或日志。已登录设备复用身份和标签；不强制重新认证。现有 legacy、其他 Site ID 或无效配置拒绝执行，人工处理后重试。SSH 超时后设备可能已部分生效，报告 unknown/failed，不能自动回滚注册或释放 Site ID。多设备不提供事务回滚。读取/应用均再次核对 SN。
## 验收与验证
隔离测试验证分配稳定、清单冲突、SN 不匹配、预检失败零写入、dry-run 不创建密钥、已有身份不创建密钥、密钥仅走 stdin、失败脱敏和汇总退出码。真实安装/注册/路由批准/维护页面访问待明确设备清单后试点。SN 是稳定资产标识，host 是随地点变化的连接参数；用途变更会改变 MagicDNS 名称。
scripts/tailscale/batch-inventory.example.json 0 → 100644
[
{
"host": "192.0.2.10",
"user": "operator",
"sn": "QARM-EXAMPLE-001",
"purpose": "coffee",
"tags": ["tag:qarm"]
},
{
"host": "192.0.2.11",
"user": "operator",
"sn": "QARM-EXAMPLE-002",
"purpose": "maintenance",
"site_id": 101,
"tags": ["tag:qarm", "tag:europe"]
}
]
scripts/tailscale/tailscale-batch-device.py 0 → 100644
#!/usr/bin/env python3
"""Linux root worker, invoked by the batch controller over SSH stdin.
Requires Python 3, Bash and an installed/running Tailscale. Not a service.
Reads one JSON request, emits one JSON response; exit 1 means failed/partial.
Auth keys exist only in memory and a mode-0600 temporary file.
"""
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
CONFIG = Path('/var/lib/tailscale-route-manager.site')
MARKER = '# Managed by qarm tailscale-route-manager site config'
TARGETS = ('192.168.11.201/32', '192.168.22.202/32')
def command(args, timeout=60):
result = subprocess.run(args, capture_output=True, text=True, timeout=timeout)
if result.returncode:
# External output can contain keys or full registration commands.
raise RuntimeError(f'{args[0]} 执行失败（退出码 {result.returncode}）')
return result.stdout
def name_for(sn, purpose):
for value in (sn, purpose):
if not isinstance(value, str) or not re.fullmatch(r'[A-Za-z0-9](?:[A-Za-z0-9-]*[A-Za-z0-9])?', value):
raise ValueError('SN 或用途格式无效')
name = f'{sn}-{purpose}'.lower()
if len(name) > 63:
raise ValueError('SN 与用途组合超过 63 字符')
return name
def probe(request):
with open(request['hw_config'], encoding='utf-8') as stream:
sn = json.load(stream)['sn']
if sn != request['sn']:
raise ValueError('设备硬件 SN 与清单不一致')
name = name_for(sn, request['purpose'])
status = json.loads(command(['tailscale', 'status', '--json']))
state = status.get('BackendState')
if state not in ('Running', 'NeedsLogin', 'NoState'):
raise ValueError(f'Tailscale 状态不支持自动注册：{state}')
mode, site = '4via6', None
if CONFIG.exists():
lines = CONFIG.read_text().splitlines()
if MARKER not in lines:
raise ValueError('Site ID 配置不是管理器创建的文件')
fields = dict(line.split('=', 1) for line in lines if '=' in line)
mode = fields.get('ROUTE_MODE', '4via6')
site = fields.get('SITE_ID')
if site is not None and (not site.isdecimal() or not 1 <= int(site) <= 65535):
raise ValueError('设备已有 Site ID 配置无效')
if mode != '4via6' or (site is not None and int(site) != request['site_id']):
raise ValueError('已有路由模式或 Site ID 与清单冲突，请先人工处理')
prefs = json.loads(command(['tailscale', 'debug', 'prefs']))
if any(route in ('192.168.11.0/24', '192.168.22.0/24')
for route in prefs.get('AdvertiseRoutes') or []):
raise ValueError('设备仍在发布旧 IPv4 路由，请先人工迁移')
if not isinstance(request['site_id'], int) or isinstance(request['site_id'], bool) or not 1 <= request['site_id'] <= 65535:
raise ValueError('Site ID 无效')
for target in TARGETS:
command(['tailscale', 'debug', 'via', str(request['site_id']), target])
command(['bash', '--version'])
return {'sn': sn, 'hostname': name, 'connected': state == 'Running',
'site_id': request['site_id'], 'previous_name': status.get('Self', {}).get('HostName')}
def apply(request):
before = probe(request)
if not before['connected']:
key = request.get('auth_key')
if not isinstance(key, str) or not key.startswith('tskey-'):
raise ValueError('未登录设备需要单次 Auth Key')
key_fd, key_path = tempfile.mkstemp(prefix='qarm-batch-auth-')
try:
with os.fdopen(key_fd, 'w') as stream:
stream.write(key)
command(['tailscale', 'up', '--timeout=60s', f'--auth-key=file:{key_path}',
f"--hostname={before['hostname']}",
'--advertise-tags=' + ','.join(request['tags'])], timeout=75)
finally:
os.unlink(key_path)
else:
command(['tailscale', 'set', '--hostname=' + before['hostname']])
# Source the trusted local deployment artifact, overriding only the UI.
# This keeps route merging and forwarding recovery in the existing manager.
with tempfile.TemporaryDirectory(prefix='qarm-batch-manager-') as directory:
manager = Path(directory) / 'manager.sh'
manager.write_text(request['manager'], encoding='utf-8')
script = '''source "$1"
message_box() { printf '%s\\n' "$1" >&2; }
input_value() { return 1; }
show_tui_progress() { :; }
configure_route_mode 4via6 || exit 1
configure_site_id "$2" || exit 1
if [[ "$3" == true ]]; then enable_routes || exit 1; fi
'''
command(['bash', '-c', script, 'batch', str(manager), str(request['site_id']),
str(request['enable_routes']).lower()], timeout=90)
after = probe(request)
if not after['connected']:
raise RuntimeError('注册后未连接')
status = json.loads(command(['tailscale', 'status', '--json']))
if status.get('Self', {}).get('HostName', '').lower() != before['hostname']:
raise RuntimeError('注册后网络名称未确认')
prefs = json.loads(command(['tailscale', 'debug', 'prefs']))
routes = prefs.get('AdvertiseRoutes') or []
if request['enable_routes']:
wanted = [command(['tailscale', 'debug', 'via', str(request['site_id']), target]).strip().splitlines()[-1]
for target in TARGETS]
if not all(route in routes for route in wanted):
raise RuntimeError('路由发布未确认')
after.update({'tailscale_ips': status.get('Self', {}).get('TailscaleIPs', []),
'dns_name': status.get('Self', {}).get('DNSName', ''),
'advertised_routes': routes,
'identity': 'reused' if before['connected'] else 'registered'})
return after
def main():
try:
request = json.load(sys.stdin)
if os.geteuid() != 0:
raise RuntimeError('需要非交互 root 权限（sudo -n）')
result = apply(request) if request['phase'] == 'apply' else probe(request)
print(json.dumps({'ok': True, **result}, ensure_ascii=False))
return 0
except Exception as exc:
# Never serialize external command output or the request payload.
message = str(exc) if isinstance(exc, (ValueError, RuntimeError)) else type(exc).__name__
print(json.dumps({'ok': False, 'error': message}, ensure_ascii=False))
return 1
if __name__ == '__main__':
sys.exit(main())
scripts/tailscale/tailscale-batch-register.py 0 → 100644
#!/usr/bin/env python3
"""Batch QArm provisioning over trusted LAN SSH (Windows/Linux Python 3).
Default is read-only preflight; --apply reserves IDs and changes devices.
Requires ssh, known host keys and sudo -n on Linux targets. OAuth credentials
stay in controller environment. Exit: 0 all succeeded, 1 failed/partial,
2 invalid arguments/inventory. See docs/tailscale/batch-registration.md.
"""
import argparse
from contextlib import contextmanager
import importlib.util
import json
import os
from pathlib import Path
import re
import shlex
import subprocess
import sys
import tempfile
HERE = Path(__file__).resolve().parent
def load_module(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
DEVICE = load_module('batch_device', HERE / 'tailscale-batch-device.py')
OAUTH = load_module('batch_oauth', HERE / 'tailscale-oauth-helper.py')
def validate_inventory(value):
if not isinstance(value, list) or not value:
raise ValueError('设备清单必须为非空 JSON 数组')
rows = []
seen_sn, seen_host = set(), set()
for row in value:
if not isinstance(row, dict):
raise ValueError('设备条目必须为对象')
allowed = {'host', 'user', 'sn', 'purpose', 'site_id', 'tags', 'port', 'hw_config'}
if set(row) - allowed:
raise ValueError('清单含未知字段')
entry = dict(row)
DEVICE.name_for(entry['sn'], entry['purpose'])
for field in ('host', 'user'):
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]*', entry[field]):
raise ValueError('SSH host 或 user 无效')
if entry['sn'].lower() in seen_sn or entry['host'].lower() in seen_host:
raise ValueError('清单存在重复 SN 或连接地址')
seen_sn.add(entry['sn'].lower())
seen_host.add(entry['host'].lower())
entry.setdefault('port', 22)
if type(entry['port']) is not int or not 1 <= entry['port'] <= 65535:
raise ValueError('SSH port 无效')
entry.setdefault('hw_config', '/opt/armaster/data/hw.json')
if not isinstance(entry['hw_config'], str) or not entry['hw_config'].startswith('/'):
raise ValueError('hw_config 必须为设备端绝对路径')
tags = entry.setdefault('tags', ['tag:qarm'])
if not isinstance(tags, list) or not tags or any(
not isinstance(tag, str) or not re.fullmatch(r'tag:[a-z0-9][a-z0-9-]*', tag) for tag in tags):
raise ValueError('tags 必须为非空小写标签列表')
rows.append(entry)
return rows
def allocate(rows, registry):
if not isinstance(registry, dict):
raise ValueError('注册表必须为 SN 到 Site ID 的 JSON 对象')
result = dict(registry)
used = set()
if len({sn.lower() for sn in result}) != len(result):
raise ValueError('注册表 SN 大小写冲突')
for sn, site in result.items():
DEVICE.name_for(sn, 'qarm')
if type(site) is not int or not 1 <= site <= 65535 or site in used:
raise ValueError('注册表存在非法或重复 Site ID')
used.add(site)
# Reserve explicit IDs before automatic allocation.
for row in rows:
existing_sn = next((sn for sn in result if sn.lower() == row['sn'].lower()), row['sn'])
if existing_sn != row['sn']:
raise ValueError('清单 SN 大小写必须与注册表一致')
site = row.get('site_id', result.get(existing_sn))
if site is None:
continue
if type(site) is not int or not 1 <= site <= 65535:
raise ValueError('Site ID 必须为 1-65535 整数')
if existing_sn in result and result[existing_sn] != site:
raise ValueError('清单与注册表的 Site ID 冲突')
if existing_sn not in result and site in used:
raise ValueError('Site ID 已被其他硬件 SN 占用')
result[existing_sn] = site
used.add(site)
for row in rows:
if row['sn'] not in result:
site = next((n for n in range(1, 65536) if n not in used), None)
if site is None:
raise ValueError('Site ID 已耗尽')
result[row['sn']] = site
used.add(site)
row['site_id'] = result[row['sn']]
return result
def write_json(path, value):
path = Path(path)
descriptor, temporary = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent)
try:
with os.fdopen(descriptor, 'w', encoding='utf-8') as stream:
json.dump(value, stream, ensure_ascii=False, indent=2)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
@contextmanager
def registry_lock(path, apply):
lock = Path(str(path) + '.lock')
if not apply:
yield
return
# Exclusive create: a crashed controller leaves a lock for manual review.
descriptor = os.open(lock, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
try:
os.close(descriptor)
yield
finally:
lock.unlink()
def remote(row, phase, args, auth_key=None):
request = {**row, 'phase': phase, 'enable_routes': args.enable_routes}
if phase == 'apply':
request['manager'] = (HERE / 'tailscale-route-manager.sh').read_text(encoding='utf-8')
request['auth_key'] = auth_key
worker = (HERE / 'tailscale-batch-device.py').read_text(encoding='utf-8')
remote_command = 'sudo -n python3 -c ' + shlex.quote(worker)
argv = ['ssh', '-T', '-o', 'BatchMode=yes', '-o', 'StrictHostKeyChecking=yes',
'-o', 'ConnectTimeout=10', '-o', 'ServerAliveInterval=15',
'-o', 'ServerAliveCountMax=2', '-p', str(row['port'])]
if args.identity_file:
argv += ['-i', str(args.identity_file)]
argv += [f"{row['user']}@{row['host']}", remote_command]
try:
completed = subprocess.run(argv, input=json.dumps(request), capture_output=True,
text=True, encoding='utf-8', timeout=240)
response = json.loads(completed.stdout)
if not isinstance(response, dict):
raise ValueError('invalid response')
if completed.returncode and response.get('ok'):
raise ValueError('inconsistent response')
return response
except subprocess.TimeoutExpired:
return {'ok': False, 'error': 'SSH 超时；设备可能已部分生效，请预检后重试'}
except (OSError, ValueError):
return {'ok': False, 'error': 'SSH 连接失败或设备响应无效；检查 host key、SSH 密钥、sudo -n 和 Python 3'}
def provision(rows, registry_path, args, transport=remote):
registry_path = Path(registry_path)
with registry_lock(registry_path, args.apply):
registry = json.loads(registry_path.read_text(encoding='utf-8-sig')) if registry_path.exists() else {}
reserved = allocate(rows, registry)
results = []
for row in rows:
result = transport(row, 'probe', args)
results.append({'sn': row['sn'], 'host': row['host'], 'phase': 'probe', **result})
if not args.apply or not all(result.get('ok') for result in results):
return results
# Reserve before obtaining keys or modifying any target. Failed devices
# keep their IDs so retries cannot allocate these IDs to other hardware.
write_json(registry_path, reserved)
for row, result in zip(rows, results):
key = None
try:
if not result['connected']:
client_id, secret = OAUTH.require_environment(os.environ)
token = OAUTH.fetch_access_token(OAUTH.API_BASE_URL, client_id, secret)
key = OAUTH.create_auth_key(OAUTH.API_BASE_URL,
os.environ.get('TS_TAILNET', '-'), token, row['tags'],
row['sn'] + '-registration', 600)
response = transport(row, 'apply', args, auth_key=key)
result.update({'phase': 'apply', **response})
except Exception:
# Helper or transport exceptions can include credentials.
result.update({'phase': 'apply', 'ok': False,
'error': '注册执行失败；检查控制端 OAuth 凭证、权限和网络，保留编号供重试'})
finally:
key = None
return results
def main():
if hasattr(sys.stdout, 'reconfigure'):
sys.stdout.reconfigure(encoding='utf-8')
parser = argparse.ArgumentParser(description='QArm 局域网 SSH 批量注册；默认只读预检')
parser.add_argument('--inventory', required=True, type=Path)
parser.add_argument('--registry', required=True, type=Path, help='持久 SN → Site ID 注册表')
modes = parser.add_mutually_exclusive_group()
modes.add_argument('--apply', action='store_true', help='执行注册、改名和 Site ID 配置')
modes.add_argument('--dry-run', action='store_true', help='只读远端预检（默认）')
parser.add_argument('--enable-routes', action='store_true', help='额外启用 4via6 路由和转发')
parser.add_argument('--identity-file', type=Path)
parser.add_argument('--report', type=Path, help='可选 JSON 汇总，不包含凭证')
args = parser.parse_args()
try:
protected = {args.inventory.resolve(), args.registry.resolve()}
if len(protected) != 2 or (args.report and args.report.resolve() in protected):
raise ValueError('清单、注册表和报告必须使用不同路径')
rows = validate_inventory(json.loads(args.inventory.read_text(encoding='utf-8-sig')))
results = provision(rows, args.registry, args)
if args.report:
write_json(args.report, results)
print(json.dumps(results, ensure_ascii=False, indent=2))
return 0 if all(result.get('ok') for result in results) else 1
except (OSError, ValueError, KeyError, TypeError) as exc:
print(f'清单或注册表无效：{type(exc).__name__}；检查路径、字段和锁文件。', file=sys.stderr)
return 2
if __name__ == '__main__':
sys.exit(main())
scripts/tailscale/tailscale-route-manager.sh
......@@ -11,7 +11,7 @@ set -euo pipefail
# 首次启用 forwarding 前记录的内核状态。
readonly INSTALL_URL="https://tailscale.com/install.sh"
readonly SCRIPT_VERSION="0.4.2"
readonly SCRIPT_VERSION="0.4.5"
readonly RC_CANCELLED=10
readonly SYSCTL_FILE="/etc/sysctl.d/99-tailscale-routes.conf"
readonly STATE_FILE="/var/lib/tailscale-route-manager.state"
......@@ -111,6 +111,83 @@ show_tui_progress() {
fi
}
capture_with_tui_progress() (
# A gauge keeps newt/dialog alive until the producer closes its input.
# An infobox exits immediately and newt restores the underlying terminal.
local title="$1"
local message="$2"
shift 2
local backend=""
if command_exists dialog; then
backend="dialog"
elif command_exists whiptail; then
backend="whiptail"
else
"$@" 2>&1
return $?
fi
local workdir
workdir="$(mktemp -d /tmp/qarm-tui-progress.XXXXXX)" || return 1
trap 'rm -f -- "$workdir/output" "$workdir/result"; rmdir -- "$workdir"' EXIT
local result=0
{
printf '0\n'
"$@" >"$workdir/output" 2>&1 || result=$?
printf '%s\n' "$result" >"$workdir/result"
printf '100\n'
} | "$backend" --title "$title" --gauge "$message" 8 72 0 >&2 || true
[[ -f "$workdir/result" ]] || return 1
cat "$workdir/output"
result="$(cat "$workdir/result")"
return "$result"
)
begin_tui_screen() {
TUI_SCREEN_DIR=""
TUI_ORIGINAL_TERM="${TERM:-dumb}"
TUI_ORIGINAL_TERMINFO="${TERMINFO-}"
TUI_HAD_TERMINFO="${TERMINFO+x}"
if ! command_exists dialog && ! command_exists whiptail; then
return 0
fi
if ! command_exists tic || ! command_exists tput; then
return 0
fi
[[ "$TUI_ORIGINAL_TERM" =~ ^[A-Za-z0-9_.+-]+$ ]] || return 0
local directory
directory="$(mktemp -d /tmp/qarm-tui-screen.XXXXXX)" || return 1
# Retain the real terminal's keys, colours and dimensions. Only suppress
# each widget's alternate-screen enter/leave; the menu owns that lifecycle.
printf 'qarm-tui|QArm persistent TUI,\n smcup@, rmcup@, use=%s,\n' "$TUI_ORIGINAL_TERM" >"$directory/source"
if ! tic -x -o "$directory" "$directory/source" 2>/dev/null; then
rm -f -- "$directory/source"
rmdir -- "$directory" 2>/dev/null || true
return 0
fi
TUI_SCREEN_DIR="$directory"
tput smcup 2>/dev/null || true
export TERMINFO="$directory"
export TERM="qarm-tui"
}
end_tui_screen() {
[[ -n "${TUI_SCREEN_DIR:-}" ]] || return 0
export TERM="$TUI_ORIGINAL_TERM"
if [[ -n "$TUI_HAD_TERMINFO" ]]; then
export TERMINFO="$TUI_ORIGINAL_TERMINFO"
else
unset TERMINFO
fi
tput rmcup 2>/dev/null || true
# Only remove the private directory produced by mktemp above.
if [[ "$TUI_SCREEN_DIR" == /tmp/qarm-tui-screen.* ]]; then
rm -rf -- "$TUI_SCREEN_DIR"
fi
}
yes_no() {
local title="$1"
local message="$2"
......@@ -377,6 +454,59 @@ get_tailscale_network_name() {
| head -n 1
}
configure_tailscale_hostname() {
local purpose="${1:-}"
if ! is_tailscale_installed; then
message_box "无法修改网络名称" "Tailscale 尚未安装。"
return 1
fi
if ! is_tailscale_logged_in; then
message_box "无法修改网络名称" "当前设备尚未登录 Tailscale。"
return 1
fi
local prefix
if ! prefix="$(hardware_sn_prefix 2>/dev/null)" || [[ -z "$prefix" ]]; then
message_box "无法修改网络名称" "无法从 ${QARM_HW_CONFIG_FILE:-$DEFAULT_HARDWARE_CONFIG_FILE} 读取有效硬件 SN。\n\n网络名称必须使用 <硬件SN>-<用途>，本次操作已停止。"
return 1
fi
local current
current="$(get_tailscale_network_name 2>/dev/null || true)"
local current_purpose=""
if [[ "${current,,}" == "${prefix,,}"* ]]; then
current_purpose="${current:${#prefix}}"
fi
if [[ -z "$purpose" ]]; then
purpose="$(input_value "修改网络名称" "硬件 SN 前缀：$prefix\n\n请输入设备用途，脚本会生成 <硬件SN>-<用途>。\n这会同时修改 MagicDNS 名称，但不会修改 Linux 系统 hostname。" "$current_purpose")" || return "$RC_CANCELLED"
fi
local requested
if ! requested="$(hardware_device_name "$purpose")"; then
message_box "用途无效" "用途必须以字母或数字开头和结尾，只能包含字母、数字和连字符；完整名称不得超过 63 个字符。"
return 1
fi
if [[ "${requested,,}" == "${current,,}" ]]; then
message_box "网络名称" "当前网络名称已经是：$current"
return 0
fi
if ! tailscale set --hostname="$requested"; then
message_box "修改失败" "Tailscale 未能更新设备网络名称。"
return 1
fi
local updated
updated="$(get_tailscale_network_name 2>/dev/null || true)"
if [[ "${updated,,}" != "${requested,,}" ]]; then
message_box "修改未确认" "命令已执行，但当前读取到的网络名称为：${updated:-未知}\n\n请刷新状态后再次确认。"
return 1
fi
message_box "网络名称已更新" "原名称：${current:-未知}\n新名称：$updated\n\nMagicDNS 名称也会随之更新；Linux 系统 hostname 未修改。"
}
is_tailscale_logged_in() {
[[ "$(get_backend_state || true)" == "Running" ]]
}
......@@ -428,13 +558,15 @@ print(f"{serial}-")
PY
}
default_device_name() {
hardware_device_name() {
local purpose="$1"
local prefix
if prefix="$(hardware_sn_prefix 2>/dev/null)" && [[ -n "$prefix" ]]; then
printf '%s\n' "$prefix"
return 0
fi
hostname -s 2>/dev/null || hostname 2>/dev/null || true
prefix="$(hardware_sn_prefix 2>/dev/null)" || return 1
valid_device_name "$purpose" || return 1
local device_name="${prefix}${purpose}"
valid_device_name "$device_name" || return 1
printf '%s\n' "$device_name"
}
oauth_runtime_ready() {
......@@ -482,16 +614,18 @@ oauth_register_device() {
systemctl start tailscaled
fi
local default_name
default_name="$(default_device_name)"
local device_name
if ! device_name="$(input_value "设备名称" "请输入新设备在 Tailscale 中显示的名称。\n\n已尝试使用 hw.json 的 SN 作为前缀；请在末尾补充名称。\n允许字母、数字和连字符，最长 63 个字符。" "$default_name")"; then
local prefix
if ! prefix="$(hardware_sn_prefix 2>/dev/null)" || [[ -z "$prefix" ]]; then
message_box "缺少硬件 SN" "无法从 ${QARM_HW_CONFIG_FILE:-$DEFAULT_HARDWARE_CONFIG_FILE} 读取有效硬件 SN。\n\n自动注册要求设备名称使用 <硬件SN>-<用途>。"
return 1
fi
local purpose
if ! purpose="$(input_value "设备用途" "硬件 SN 前缀：$prefix\n\n请输入设备用途，脚本会生成 <硬件SN>-<用途>。\n允许字母、数字和连字符。" "")"; then
return "$RC_CANCELLED"
fi
if ! valid_device_name "$device_name"; then
message_box "名称无效" "设备名称格式无效：$device_name
名称必须以字母或数字开头和结尾，只能包含字母、数字和连字符，最长 63 个字符。"
local device_name
if ! device_name="$(hardware_device_name "$purpose")"; then
message_box "用途无效" "用途必须以字母或数字开头和结尾，只能包含字母、数字和连字符；完整名称不得超过 63 个字符。"
return 1
fi
......@@ -1457,6 +1591,11 @@ get_main_status_text() {
'
}
get_menu_snapshot() {
get_main_action
get_main_status_text "$1"
}
run_main_action() {
if ! is_tailscale_installed; then
install_tailscale
......@@ -1480,7 +1619,7 @@ show_status() {
elif [[ -t 1 ]] && [[ -z "${NO_COLOR:-}" ]] && [[ "${TERM:-}" != "dumb" ]]; then
mode="ansi"
fi
output_box "Tailscale 路由状态" "$(get_status_text "$mode")"
output_box "Tailscale 路由状态" "$(capture_with_tui_progress "Tailscale 路由状态" "正在读取状态，请稍候……" get_status_text "$mode")"
}
run_netcheck() {
......@@ -1496,7 +1635,7 @@ show_netcheck() {
local output
local result=0
output="$(run_netcheck 2>&1)" || result=$?
output="$(capture_with_tui_progress "Tailscale 网络检测" "正在执行网络检测，请稍候……" run_netcheck)" || result=$?
[[ -z "$output" ]] && output="网络检测没有返回内容。"
if [[ "$result" -ne 0 ]]; then
output="$output
......@@ -1520,18 +1659,26 @@ run_tui_branch() {
return 0
}
run_menu() {
run_menu() (
# Scope TERM/TERMINFO to this menu process, including every child widget.
begin_tui_screen
trap 'end_tui_screen; cleanup' EXIT
while true; do
local action
local choice
action="$(get_main_action)"
local snapshot
local status_text
if command_exists dialog; then
snapshot="$(capture_with_tui_progress "Tailscale 路由管理器" "正在刷新主页面，请稍候……" get_menu_snapshot dialog)"
action="${snapshot%%$'\n'*}"
status_text="${snapshot#*$'\n'}"
if ! choice="$(dialog --colors --stdout --no-cancel --title "Tailscale 路由管理器" \
--menu "$(get_main_status_text dialog)\n\n请选择操作：" 29 84 12 \
--menu "$status_text\n\n请选择操作：" 30 84 13 \
"toggle" "$action" \
"route-mode" "选择路由模式" \
"site-id" "配置 4via6 Site ID" \
"hostname" "修改 Tailscale 网络名称" \
"login" "登录 Tailscale" \
"logout" "退出登录" \
"register" "OAuth 注册新设备" \
......@@ -1542,11 +1689,15 @@ run_menu() {
continue
fi
elif command_exists whiptail; then
snapshot="$(capture_with_tui_progress "Tailscale 路由管理器" "正在刷新主页面，请稍候……" get_menu_snapshot whiptail)"
action="${snapshot%%$'\n'*}"
status_text="${snapshot#*$'\n'}"
if ! choice="$(whiptail --nocancel --title "Tailscale 路由管理器" \
--menu "$(get_main_status_text whiptail)\n\n请选择操作：" 29 84 12 \
--menu "$status_text\n\n请选择操作：" 30 84 13 \
"toggle" "$action" \
"route-mode" "选择路由模式" \
"site-id" "配置 4via6 Site ID" \
"hostname" "修改 Tailscale 网络名称" \
"login" "登录 Tailscale" \
"logout" "退出登录" \
"register" "OAuth 注册新设备" \
......@@ -1557,23 +1708,25 @@ run_menu() {
continue
fi
else
action="$(get_main_action)"
clear
get_status_text
printf '\n============================================================\n'
printf '1) %s\n2) 选择路由模式\n3) 配置 4via6 Site ID\n4) 登录 Tailscale\n5) 退出登录\n6) OAuth 注册新设备\n7) 刷新/查看状态\n8) 网络检测\n9) 安装/检查 Tailscale\n10) 退出\n' "$action"
printf '1) %s\n2) 选择路由模式\n3) 配置 4via6 Site ID\n4) 修改 Tailscale 网络名称\n5) 登录 Tailscale\n6) 退出登录\n7) OAuth 注册新设备\n8) 刷新/查看状态\n9) 网络检测\n10) 安装/检查 Tailscale\n11) 退出\n' "$action"
printf '============================================================\n'
read -r -p "请选择 [1-10]: " choice
read -r -p "请选择 [1-11]: " choice
case "$choice" in
1) choice="toggle" ;;
2) choice="route-mode" ;;
3) choice="site-id" ;;
4) choice="login" ;;
5) choice="logout" ;;
6) choice="register" ;;
7) choice="status" ;;
8) choice="netcheck" ;;
9) choice="install" ;;
10) choice="exit" ;;
4) choice="hostname" ;;
5) choice="login" ;;
6) choice="logout" ;;
7) choice="register" ;;
8) choice="status" ;;
9) choice="netcheck" ;;
10) choice="install" ;;
11) choice="exit" ;;
*) continue ;;
esac
fi
......@@ -1588,6 +1741,9 @@ run_menu() {
site-id)
run_tui_branch "正在打开 Site ID 设置……" configure_site_id
;;
hostname)
run_tui_branch "正在读取 Tailscale 网络名称……" configure_tailscale_hostname
;;
login)
run_tui_branch "正在检查 Tailscale 登录状态……" login_from_menu
;;
......@@ -1611,7 +1767,7 @@ run_menu() {
esac
done
clear || true
}
)
show_help() {
cat <<EOF
......@@ -1622,9 +1778,10 @@ show_help() {
install 安装或检查 Tailscale
route-mode 配置路由模式（4via6 或 legacy）
site-id 配置本设备唯一的 4via6 Site ID（1-65535）
hostname 修改设备的 Tailscale 网络名称和 MagicDNS 名称
login 登录 Tailscale
logout 退出 Tailscale，并使当前登录凭据失效
register 通过 OAuth 选择设备名称和标签并自动注册
register 通过 OAuth 输入用途、组合硬件 SN、选择标签并自动注册
enable 登录并发布本脚本管理的子网路由
disable 停止发布本脚本管理的子网路由并恢复 forwarding
status 输出状态，不修改系统
......@@ -1654,6 +1811,10 @@ main() {
require_root
configure_route_mode "${2:-}"
;;
hostname)
require_root
configure_tailscale_hostname "${2:-}"
;;
menu|install|login|logout|register|enable|disable)
require_root
case "$action" in
......
tests/tailscale/test-route-manager.sh
......@@ -28,10 +28,15 @@ QARM_HW_CONFIG_FILE="$hw_file"
export QARM_HW_CONFIG_FILE
[[ "$(hardware_sn_prefix)" == "CA2EUP-20261020-001-" ]] \
|| fail "hardware serial prefix was not loaded"
[[ "$(hardware_device_name maintenance)" == "CA2EUP-20261020-001-maintenance" ]] \
|| fail "hardware serial and purpose did not form the device name"
printf '%s\n' '{"sn":"invalid_sn"}' >"$hw_file"
if hardware_sn_prefix >/dev/null 2>&1; then
fail "invalid hardware serial was accepted"
fi
if hardware_device_name qarm >/dev/null 2>&1; then
fail "device name was generated without a valid hardware serial"
fi
printf '%s\n' '{invalid json' >"$hw_file"
if hardware_sn_prefix >/dev/null 2>&1; then
fail "invalid hardware JSON was accepted"
......@@ -44,9 +49,14 @@ helper_path="$(find_oauth_helper)"
TAILSCALE_SET_ARGUMENT=""
MOCK_ADVERTISED_ROUTES='"192.168.11.0/24","fd7a:115c:a1e0:b1a:0:65:c0a8:bc9/128","fd7a:115c:a1e0:b1a:0:65:c0a8:16ca/128","10.20.30.0/24"'
MOCK_NETWORK_NAME="CA2EUP-20261020-001-coffee"
tailscale() {
if [[ "${1:-}" == "set" ]]; then
TAILSCALE_SET_ARGUMENT="${2:-}"
if [[ "${2:-}" == --hostname=* ]]; then
MOCK_NETWORK_NAME="${2#--hostname=}"
MOCK_NETWORK_NAME="${MOCK_NETWORK_NAME,,}"
fi
return 0
fi
if [[ "${1:-} ${2:-}" == "debug via" ]]; then
......@@ -59,11 +69,11 @@ tailscale() {
fi
case "${1:-} ${2:-}" in
"status --json")
cat <<'JSON'
cat <<JSON
{
"BackendState": "Running",
"Self": {
"HostName": "CA2EUP-20261020-001-coffee"
"HostName": "$MOCK_NETWORK_NAME"
},
"Peer": {
"peer-id": {
......@@ -172,6 +182,12 @@ command_exists() {
return 1
}
progress_failure_probe() { printf 'captured failure\n'; return 7; }
progress_result=0
progress_output="$(capture_with_tui_progress "测试" "处理中" progress_failure_probe)" || progress_result=$?
[[ "$progress_output" == "captured failure" && "$progress_result" -eq 7 ]] \
|| fail "ordinary terminal progress capture lost command output or exit status"
mode="$(printf '1\n' | select_route_mode)"
[[ "$mode" == "4via6" ]] || fail "recommended 4via6 route mode was not selected"
mode="$(printf '2\n' | select_route_mode)"
......@@ -188,6 +204,21 @@ message_box() { return 0; }
enable_forwarding() { FORWARDING_CALLS=$((FORWARDING_CALLS + 1)); }
restore_forwarding() { return 0; }
TAILSCALE_SET_ARGUMENT=""
configure_tailscale_hostname "maintenance" \
|| fail "Tailscale network name update failed"
[[ "$TAILSCALE_SET_ARGUMENT" == "--hostname=CA2EUP-20261020-001-maintenance" ]] \
|| fail "network name update did not use tailscale set --hostname"
[[ "$(get_tailscale_network_name)" == "ca2eup-20261020-001-maintenance" ]] \
|| fail "updated Tailscale network name was not verified"
TAILSCALE_SET_ARGUMENT=""
configure_tailscale_hostname "maintenance" \
|| fail "unchanged network name should be successful"
[[ -z "$TAILSCALE_SET_ARGUMENT" ]] || fail "unchanged network name was written again"
if configure_tailscale_hostname "invalid_name"; then
fail "invalid Tailscale network name was accepted"
fi
if configure_route_mode legacy; then
fail "route mode changed while current 4via6 routes were still advertised"
fi
......@@ -329,6 +360,7 @@ CAPTURED_AUTH_FILE=""
is_tailscale_installed() { return 0; }
is_tailscale_logged_in() { [[ "$REGISTERED" -eq 1 ]]; }
oauth_runtime_ready() { return 0; }
hardware_sn_prefix() { printf '%s\n' "CA2EUP-20261020-001-"; }
input_value() { printf '%s\n' "qarm-001"; }
find_oauth_helper() { printf '%s\n' "/mock/oauth-helper.py"; }
select_tags() { printf '%s\n' "tag:germany" "tag:qarm-router"; }
......@@ -353,7 +385,7 @@ tailscale() {
for argument in "$@"; do
case "$argument" in
--auth-key=file:*) auth_file="${argument#--auth-key=file:}" ;;
--hostname=qarm-001) found_hostname=1 ;;
--hostname=CA2EUP-20261020-001-qarm-001) found_hostname=1 ;;
--advertise-tags=tag:germany,tag:qarm-router) found_tags=1 ;;
esac
done
......
tests/tailscale/test_batch_registration.py 0 → 100644
import argparse
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
from unittest.mock import patch
SCRIPTS = Path(__file__).resolve().parents[2] / 'scripts' / 'tailscale'
SPEC = importlib.util.spec_from_file_location('batch', SCRIPTS / 'tailscale-batch-register.py')
BATCH = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(BATCH)
DEVICE = BATCH.DEVICE
def inventory():
return [{'host': '192.0.2.10', 'user': 'operator', 'sn': 'QARM-001', 'purpose': 'coffee'}]
def options(**kwargs):
return argparse.Namespace(apply=True, enable_routes=False, identity_file=None, **kwargs)
class BatchRegistrationTests(unittest.TestCase):
def test_allocation_reserves_explicit_ids_before_automatic_ids(self):
rows = BATCH.validate_inventory(inventory() + [
{'host': '192.0.2.11', 'user': 'operator', 'sn': 'QARM-002',
'purpose': 'coffee', 'site_id': 1}])
self.assertEqual({'QARM-002': 1, 'QARM-001': 2}, BATCH.allocate(rows, {}))
def test_relocation_reuses_sn_allocation(self):
rows = BATCH.validate_inventory(inventory())
rows[0]['host'] = '192.0.2.99'
self.assertEqual({'QARM-001': 101}, BATCH.allocate(rows, {'QARM-001': 101}))
self.assertEqual(101, rows[0]['site_id'])
def test_conflicting_and_duplicate_ids_rejected(self):
for registry in ({'QARM-001': True}, {'QARM-001': 0},
{'QARM-001': 1, 'QARM-002': 1}):
with self.assertRaises(ValueError):
BATCH.allocate(BATCH.validate_inventory(inventory()), registry)
rows = BATCH.validate_inventory(inventory())
rows[0]['site_id'] = 2
with self.assertRaises(ValueError):
BATCH.allocate(rows, {'QARM-001': 1})
def test_invalid_inputs_rejected_before_ssh(self):
for field, value in [('host', '-oProxyCommand=bad'), ('user', 'a;id'),
('tags', ['tag:Bad']), ('purpose', 'bad_name')]:
rows = inventory()
rows[0][field] = value
with self.assertRaises(ValueError):
BATCH.validate_inventory(rows)
with self.assertRaises(ValueError):
BATCH.validate_inventory(inventory() * 2)
def test_dry_run_never_writes_or_obtains_keys(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
calls = []
def transport(row, phase, args):
calls.append(phase)
return {'ok': True, 'connected': False}
with patch.object(BATCH.OAUTH, 'fetch_access_token') as token:
args = options()
args.apply = False
BATCH.provision(BATCH.validate_inventory(inventory()), registry, args, transport)
token.assert_not_called()
self.assertEqual(['probe'], calls)
self.assertEqual([], list(Path(directory).iterdir()))
def test_failed_probe_prevents_every_write(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
with patch.object(BATCH.OAUTH, 'fetch_access_token') as token:
results = BATCH.provision(BATCH.validate_inventory(inventory()), registry,
options(), lambda *args: {'ok': False, 'error': 'SN mismatch'})
token.assert_not_called()
self.assertFalse(results[0]['ok'])
self.assertFalse(registry.exists())
def test_registered_device_reuses_identity_and_reserves_id(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
def transport(row, phase, args, auth_key=None):
self.assertIsNone(auth_key)
return {'ok': True, 'connected': True}
with patch.object(BATCH.OAUTH, 'fetch_access_token') as token:
results = BATCH.provision(BATCH.validate_inventory(inventory()), registry,
options(), transport)
token.assert_not_called()
self.assertTrue(results[0]['ok'])
self.assertEqual({'QARM-001': 1}, json.loads(registry.read_text()))
def test_failed_registration_preserves_reservation_and_hides_secret(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
def transport(row, phase, args, auth_key=None):
if phase == 'apply':
raise RuntimeError('tskey-supersecret')
return {'ok': True, 'connected': False}
with patch.object(BATCH.OAUTH, 'require_environment', return_value=('id', 'secret')), \
patch.object(BATCH.OAUTH, 'fetch_access_token', return_value='token'), \
patch.object(BATCH.OAUTH, 'create_auth_key', return_value='tskey-supersecret'):
results = BATCH.provision(BATCH.validate_inventory(inventory()), registry,
options(), transport)
self.assertFalse(results[0]['ok'])
self.assertNotIn('supersecret', json.dumps(results))
self.assertEqual({'QARM-001': 1}, json.loads(registry.read_text()))
def test_registry_lock_blocks_second_writer(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
with BATCH.registry_lock(registry, True):
with self.assertRaises(FileExistsError):
with BATCH.registry_lock(registry, True):
self.fail('second writer acquired lock')
def test_new_device_receives_one_key_and_success_report_has_no_key(self):
with tempfile.TemporaryDirectory() as directory:
registry = Path(directory) / 'sites.json'
phases = []
def transport(row, phase, args, auth_key=None):
phases.append(phase)
if phase == 'apply':
self.assertEqual('tskey-single-use', auth_key)
self.assertTrue(registry.exists())
return {'ok': True, 'connected': True, 'identity': 'registered'}
return {'ok': True, 'connected': False}
with patch.object(BATCH.OAUTH, 'require_environment', return_value=('id', 'secret')), \
patch.object(BATCH.OAUTH, 'fetch_access_token', return_value='token'), \
patch.object(BATCH.OAUTH, 'create_auth_key', return_value='tskey-single-use') as create:
results = BATCH.provision(BATCH.validate_inventory(inventory()), registry,
options(), transport)
create.assert_called_once()
self.assertEqual(600, create.call_args.args[-1])
self.assertEqual(['probe', 'apply'], phases)
self.assertTrue(results[0]['ok'])
self.assertNotIn('tskey-', json.dumps(results))
def test_one_failed_preflight_stops_entire_batch(self):
with tempfile.TemporaryDirectory() as directory:
rows = BATCH.validate_inventory(inventory() + [
{'host': '192.0.2.11', 'user': 'operator', 'sn': 'QARM-002', 'purpose': 'coffee'}])
def transport(row, phase, args):
self.assertEqual('probe', phase)
return {'ok': row['sn'] == 'QARM-001', 'connected': False}
results = BATCH.provision(rows, Path(directory) / 'sites.json', options(), transport)
self.assertEqual(2, len(results))
self.assertEqual([], list(Path(directory).iterdir()))
def test_existing_legacy_mode_or_different_site_is_rejected(self):
with tempfile.TemporaryDirectory() as directory:
hw = Path(directory) / 'hw.json'
hw.write_text('{"sn":"QARM-001"}')
config = Path(directory) / 'site'
for fields in ('ROUTE_MODE=legacy\nSITE_ID=1', 'ROUTE_MODE=4via6\nSITE_ID=2'):
config.write_text(DEVICE.MARKER + '\n' + fields)
with patch.object(DEVICE, 'CONFIG', config), \
patch.object(DEVICE, 'command', return_value='{"BackendState":"Running"}'):
with self.assertRaises(ValueError):
DEVICE.probe({'sn': 'QARM-001', 'purpose': 'coffee', 'site_id': 1,
'hw_config': str(hw)})
def test_successful_device_apply_reads_back_name_ips_and_routes(self):
def command(args, **kwargs):
if args[:2] == ['tailscale', 'status']:
return json.dumps({'Self': {'HostName': 'qarm-001-coffee',
'TailscaleIPs': ['100.64.0.1'], 'DNSName': 'qarm-001-coffee.example.'}})
if args[:3] == ['tailscale', 'debug', 'prefs']:
return '{"AdvertiseRoutes": ["fd00::1/128", "fd00::2/128"]}'
if args[:3] == ['tailscale', 'debug', 'via']:
return 'fd00::1/128' if args[-1] == DEVICE.TARGETS[0] else 'fd00::2/128'
return ''
with patch.object(DEVICE, 'probe', return_value={
'connected': True, 'hostname': 'qarm-001-coffee'}), \
patch.object(DEVICE, 'command', side_effect=command) as run:
result = DEVICE.apply({'manager': '# mock', 'site_id': 1, 'enable_routes': True})
self.assertEqual('reused', result['identity'])
self.assertEqual(['100.64.0.1'], result['tailscale_ips'])
self.assertFalse(any(call.args[0][:2] == ['tailscale', 'up'] for call in run.call_args_list))
def test_key_is_only_passed_in_stdin_and_host_key_checking_enabled(self):
row = BATCH.validate_inventory(inventory())[0]
row['site_id'] = 1
with patch.object(BATCH.subprocess, 'run', return_value=
subprocess.CompletedProcess([], 0, '{"ok": true}', '')) as run:
BATCH.remote(row, 'apply', options(), 'tskey-secret')
argv = run.call_args.args[0]
self.assertNotIn('tskey-secret', str(argv))
self.assertIn('tskey-secret', run.call_args.kwargs['input'])
self.assertIn('StrictHostKeyChecking=yes', argv)
self.assertIn('sudo -n python3', argv[-1])
def test_device_checks_sn_before_any_tailscale_command(self):
with tempfile.TemporaryDirectory() as directory:
hw = Path(directory) / 'hw.json'
hw.write_text('{"sn":"OTHER"}')
with patch.object(DEVICE, 'command') as run:
with self.assertRaises(ValueError):
DEVICE.probe({'sn': 'QARM-001', 'hw_config': str(hw)})
run.assert_not_called()
def test_temp_key_removed_even_when_registration_fails(self):
paths = []
def command(args, **kwargs):
if args[0:2] == ['tailscale', 'up']:
filename = next(arg.split('file:', 1)[1] for arg in args if arg.startswith('--auth-key='))
paths.append(Path(filename))
self.assertEqual('tskey-test', paths[0].read_text())
if os.name == 'posix':
self.assertEqual(0o600, paths[0].stat().st_mode & 0o777)
raise RuntimeError('failed')
with patch.object(DEVICE, 'probe', return_value={'connected': False, 'hostname': 'qarm-001-coffee'}), \
patch.object(DEVICE, 'command', side_effect=command):
with self.assertRaises(RuntimeError):
DEVICE.apply({'auth_key': 'tskey-test', 'tags': ['tag:qarm']})
self.assertTrue(paths)
self.assertFalse(paths[0].exists())
if __name__ == '__main__':
unittest.main()
tests/tailscale/test_tui_progress.py 0 → 100644
"""Linux pseudo-terminal regression: a real whiptail must cover slow jobs."""
import os
from pathlib import Path
import re
import shlex
import shutil
import subprocess
import tempfile
import unittest
MANAGER = Path(os.environ.get('QARM_TEST_MANAGER', str(
Path(__file__).resolve().parents[2] / 'scripts/tailscale/tailscale-route-manager.sh')))
@unittest.skipUnless(os.name == 'posix' and shutil.which('script') and shutil.which('whiptail'),
'requires Linux script/whiptail pseudo-terminal')
class TuiProgressTests(unittest.TestCase):
@unittest.skipUnless(shutil.which('tic') and shutil.which('tput'), 'requires terminfo tools')
def test_menu_netcheck_result_and_refresh_share_one_screen(self):
with tempfile.TemporaryDirectory() as directory:
script = Path(directory) / 'session.sh'
count = Path(directory) / 'menus'
script.write_text('''#!/usr/bin/env bash
source ''' + shlex.quote(str(MANAGER)) + '''
get_menu_snapshot() { printf '检查\\n状态正常\\n'; }
run_netcheck() { sleep 0.1; printf 'mock report\\n'; }
whiptail() {
if [[ " $* " == *" --menu "* ]]; then
printf '0\\n' | command whiptail --gauge Menu 8 72 0 >/dev/tty 2>/dev/tty
if [[ -f ''' + shlex.quote(str(count)) + ''' ]]; then
printf exit >&2
else
touch ''' + shlex.quote(str(count)) + '''
printf netcheck >&2
fi
else
command whiptail "$@"
fi
}
output_box() { printf '0\\n' | command whiptail --gauge Result 8 72 0; }
printf BEFORE_SESSION
run_menu
printf AFTER_SESSION
''')
result = subprocess.run(['script', '-q', '-c', 'bash ' + shlex.quote(str(script)), '/dev/null'],
input='', capture_output=True, text=True, timeout=10,
env={**os.environ, 'TERM': 'xterm'})
self.assertEqual(0, result.returncode, result.stderr)
screen = result.stdout.split('BEFORE_SESSION', 1)[1].split('AFTER_SESSION', 1)[0]
self.assertEqual(1, screen.count('\x1b[?1049h'), 'dialogs reopen the alternate screen')
self.assertEqual(1, screen.count('\x1b[?1049l'), 'dialogs restore the shell between steps')
def test_real_whiptail_does_not_restore_terminal_during_netcheck(self):
with tempfile.TemporaryDirectory() as directory:
script = Path(directory) / 'run.sh'
# Write markers directly to the pty, outside the captured job output.
script.write_text('''#!/usr/bin/env bash
source ''' + shlex.quote(str(MANAGER)) + '''
command_exists() { [[ "$1" == whiptail ]]; }
run_netcheck() {
sleep 0.1
printf JOB_RUNNING >/dev/tty
sleep 0.3
printf JOB_DONE >/dev/tty
printf 'Report: mock output\\n'
return 7
}
output_box() { printf 'RESULT:%s' "$2"; }
show_netcheck
''')
result = subprocess.run(['script', '-q', '-c', 'bash ' + shlex.quote(str(script)), '/dev/null'],
input='', capture_output=True, text=True, timeout=10,
env={**os.environ, 'TERM': 'xterm'})
self.assertEqual(0, result.returncode, result.stderr)
output = result.stdout
self.assertIn('JOB_RUNNING', output)
self.assertIn('JOB_DONE', output)
start = output.index('JOB_RUNNING')
end = output.index('JOB_DONE')
enter = output.rfind('\x1b[?1049h', 0, start)
leave = output.find('\x1b[?1049l', enter)
self.assertGreaterEqual(enter, 0, 'no real TUI was opened')
self.assertGreater(leave, end, 'terminal restored before the slow command finished')
self.assertIsNone(re.search(r'\x1b\[\?(?:1049|1047|47)l', output[start:end]))
self.assertIn('Report: mock output', output)
self.assertIn('7', output[output.index('RESULT:'):])
if __name__ == '__main__':
unittest.main()
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment