Yggdrasil-AI-labs/wdgwars-api-tester
GitHub: Yggdrasil-AI-labs/wdgwars-api-tester
一款仅依赖 Python 标准库的单文件工具,用于系统性探测 WDGoWars HTTP API 的端点存活状态、故障分类与持续监控告警。
Stars: 1 | Forks: 0
# wdgwars-api-tester
对 **[WDGoWars](https://wdgwars.pl/)** HTTP API 接口进行系统性探测。
构建于 2026-05-29,正值大规模 `/api/*` 404 宕机期间。这个工具的意义在于用一条命令回答那天花了整整一个小时用 curl 才搞清楚的问题:
- 哪些 endpoint 存活,哪些返回了带样式的 404 页面?
- 未认证的 `/api/me` 是返回 401(预期行为)还是 404(路由未绑定)?
- `/api/stats` 是否暴露了 LiteSpeed 管理遥测数据泄露?
- 自上次快照以来有什么变化吗?
仅使用标准库的 Python 3。无需 `pip install`。单文件。
## 家族
WDGoWars 馈送器家族中的兄弟仓库:
- [Muninn](https://github.com/HiroAlleyCat/adsb-to-wdgwars) — ADS-B 馈送器
- [Heimdall](https://github.com/HiroAlleyCat/meshcore-to-wdgwars) — MeshCore LoRa 馈送器
- [wigle-to-wdgwars](https://github.com/HiroAlleyCat/wigle-to-wdgwars) — WiGLE Wi-Fi/BLE 馈送器
- [gungnir](https://github.com/HiroAlleyCat/gungnir) — 共享的 HMAC 传输库
## 快速开始
```
# 使用全部三种 auth 变体(none、garbage、valid)探测 apex
python3 wdgwars_api_tester.py
# 添加 www. 和 api. 子域名
python3 wdgwars_api_tester.py --hosts all
# 探测自定义 host(staging、fork、local mock) — 任何以
# http:// 或 https:// 开头的内容都会成为目标,替代 wdgwars.pl。
python3 wdgwars_api_tester.py --hosts http://127.0.0.1:9999 --variants none
# Machine-readable
python3 wdgwars_api_tester.py --json > snapshot.json
# 仅输出整体 verdict 词 + exit code(适用于 shell / CI)
python3 wdgwars_api_tester.py --quiet --variants none,garbage
# → 打印 `HEALTHY` / `DEGRADED` / `OUTAGE` / `UNREACHABLE`
# 加上可选的 `+LEAK` 或 `+SENTINEL-DIVERGED` 后缀。
# 每 60s 轮询一次,在 state change 时打印紧凑的 deltas。
# Full table 会在恢复时刻(首次转换为 HEALTHY)打印。
python3 wdgwars_api_tester.py --watch 60
# Snapshot 一次,然后将后续运行与它进行 diff
python3 wdgwars_api_tester.py --baseline baseline.json
# Watch + 在 state change 时发送 Telegram 自我告警(无需 bridge)
export TELEGRAM_BOT_TOKEN=123456:ABC...
export TELEGRAM_CHAT_ID=-1001234567890
python3 wdgwars_api_tester.py --watch 60 --alert-telegram
# Watch + Discord / Slack / n8n / PagerDuty(任意 webhook URL)
python3 wdgwars_api_tester.py --watch 60 \
--alert-webhook https://discord.com/api/webhooks/.../...
# Watch + 在 state change 时执行任意 shell 命令
python3 wdgwars_api_tester.py --watch 60 \
--exec-on-change 'echo "$WDGWARS_PREV_OVERALL → $WDGWARS_OVERALL" | mail -s "wdgwars alert" me@example.com'
```
## API key
优先级与 [wigle-to-wdgwars](https://github.com/HiroAlleyCat/wigle-to-wdgwars) 相同:
1. `--key` CLI 标志
2. `$WDGWARS_API_KEY`
3. `~/.config/wigle-to-wdgwars/wdgwars.key`
如果未找到 key,将自动放弃 `valid` 变体,仅运行 `none` 和 `garbage` 变体。
## 探测内容
| Probe | Method | Path | Auth | Notes |
|---|---|---|---|---|
| `api-root` | GET | `/api/` | no | /api/ 子树的基准形态。 |
| `me` | GET | `/api/me` | yes | 身份。未认证 → 401,而非 404。 |
| `upload-history` | GET | `/api/upload-history?limit=5` | yes | 添加于 2026-04-27。 |
| `upload-csv` | POST | `/api/upload-csv` | yes | Multipart WiGLE-1.6,混合 Types。 |
| `v2-upload-csv` | POST + GET | `/api/v2/upload-csv` → `/api/v2/upload-job/` | yes | 异步 pipeline:POST 202 → 轮询直到 `done`/`failed`(间隔 1 秒轮询 6 次)。捕获独立于 v1 的 v2-parser 回归问题。 |
| `signed-upload` | GET | `/api/upload/` | yes | HMAC JSON endpoint。健康状态下 GET → 405。 |
| `me-aps` | GET | `/api/me/aps?limit=1` | yes | 调用者自己的 AP 回读(支持 `?since=` 增量同步)。 |
| `aircraft` | GET | `/api/aircraft` | yes | ADS-B 实时快照(顶级数组)。 |
| `meshcore` | GET | `/api/meshcore` | yes | MeshCore 实时快照(顶级数组)。 |
| `territories` | GET | `/api/territories` | yes | 全局帮派凸包(顶级数组)。 |
| `member-territories` | GET | `/api/member-territories` | yes | 基于网格的 cell 及帮派凸包。5 分钟快照。 |
| `leaderboard` | GET | `/api/leaderboard` | yes | 5 个排行榜。5 分钟快照。 |
| `bounties` | GET | `/api/bounties` | yes | 开放的赏金任务(最多 200 个)。由于与最初五个 handler 相同的 regex-cascade bug,从 2026-06-03 起返回 404;已于 2026-06-04 修复。 |
| `team-messages` | GET | `/api/team/messages` | yes | 调用者所在帮派的消息列表。 |
| `team-messages-id` | GET | `/api/team/messages/1` | yes | 按规范仅支持 DELETE — 2026-06-04 之后 GET → 405 + `Allow: DELETE`。健康状态由 METHOD 裁决表示。 |
| `health-asked-for` | GET | `/api/health` | no | 尚未实现。在 bug #1 中被要求添加。 |
| `stats-leak-check` | GET | `/api/stats` | no | 如果 body 带有 LSWS 管理-遥测指纹,则触发 LEAK。(locosp 于 2026-05-30 的修复已上线 — endpoint 现在会 302 跳转至 login;规则在 v0.6.1 中已收紧,改为检测内容,而不仅仅是状态。) |
| `api-sentinel-404-a/b/c` | GET | `/api/` × 3 | no | /api/ 404 页面的法定指纹(需要 3 票中的 2 票多数)。 |
| `non-api-sentinel-404` | GET | `/` | no | 为非 /api/ 404 页面生成指纹。 |
| `changelog-control` | GET | `/changelog` | no | 公共页面可达性对照组。 |
## 裁决
| Verdict | Meaning |
|---|---|
| `OK` | 2xx 响应,body 区别于任何 404 哨兵。 |
| `AUTH-REQUIRED` | 401。Endpoint 存活,且以符合规范的 JSON 格式拒绝了该 key。 |
| `AUTH-REDIRECT` | 3xx 响应,其 `Location` 指向 `/login...`。认证网关正在工作,但该 endpoint 是通过 web-session 流程连接的,而不是返回 401 JSON — 这对于 API 调用者来说属于路由形态回归,但不是安全/可用性问题。不会升级为 DEGRADED。 |
| `REDIRECT-{n}` | 3xx 响应,其 `Location` 不匹配 `/login`(通用捕获,防止意外重定向伪装成 OK)。 |
| `DEAD` | Body 哈希匹配 /api/ 404 法定哨兵。路由未绑定。 |
| `DEAD-NONAPI` | Body 匹配非 /api/ 404 哨兵。 |
| `LEAK` | Body 携带 LiteSpeed 管理-遥测指纹(`lsphp_processes` / `top_domains` / `lsphp`)。在 v0.6.1 中已泛化 — 可在任何探测中触发,不仅限于 `stats-leak-check`。从“stats 返回 200”收紧,因为一旦 locosp 于 2026-05-30 的修复上线且 `/api/stats` 开始 302 跳转至 `/login`,纯粹基于状态的规则就会产生误报。 |
| `404` | 404 响应但 body 区别于各哨兵。 |
| `METHOD` | 405。健康的 endpoint,但使用了错误的动词。 |
| `ERROR` | 网络/超时/URL 错误。 |
| `SENTINEL` | 3 个 /api/ 法定哨兵之一,与多数一致。 |
| `SENTINEL-OUTLIER` | 3 个哨兵中与其他 2 个不一致的 1 个(例如 CDN 缓存滑落)。DEAD 检测仍通过 2 票多数正常工作。 |
| `SENTINEL-DIVERGED` | 所有 3 个哨兵返回了截然不同的 body。该主机的 DEAD 检测已禁用。在信任结果之前,请排查诊断信息。 |
| `SENTINEL-NONAPI` | 非 /api/ 404 指纹探测。 |
总体状态摘要为以下之一:
- `HEALTHY` — 没有 DEAD,没有 ERROR,没有 LEAK。
- `UNREACHABLE` — 所有探测均出错。DNS 问题、无网络或主机宕机。
- `DEGRADED` — 至少有一个探测为 DEAD。
- `OUTAGE` — 携带有效 key 的 `/api/me` 为 DEAD。整个 API 接口均宕机。
- `…+LEAK` — 当 `/api/stats` 暴露时,附加在上述任何状态之后。
- `…+SENTINEL-DIVERGED` — 当 3 个法定哨兵无法就指纹达成一致时附加。受影响主机的 DEAD 检测已禁用;在信任结果前请先进行排查。
对于 DEGRADED/OUTAGE/UNREACHABLE/LEAK/SENTINEL-DIVERGED,退出代码为 `1`;对于 HEALTHY,则为 `0`。
## 按计划运行
将其放入 cron、systemd timer 或 Windows 任务计划程序中。将 `--baseline` 与 `--json` 配合使用可记录每次快照以供后续趋势分析,或者在长期运行的主机上使用 `--watch`,以便在 API 恢复时获得单一的状态更改通知。
Cron 示例:
```
*/5 * * * * cd /opt/wdgwars-api-tester && \
python3 wdgwars_api_tester.py --baseline /var/log/wdgwars/baseline.json \
--json >> /var/log/wdgwars/snapshots.jsonl
```
## 感知宕机的退避机制
当 LOCOSP 达到其记录的每日上限(UTC 午夜重置)、受到单 IP 速率限制或传输失败时,`--watch` 模式会根据不良裁决(`429` 或 `ERROR`)的比例检测到宕机,并逐步延长两次扫描之间的休眠时间,而不是以全频率强行推进。在首次完全正常的扫描后重置为正常频率。
默认:当扫描中 ≥30% 的结果为 `429`/`ERROR` 时触发。每次连续的宕机扫描休眠时间加倍(`--watch` 的 2 倍、4 倍、8 倍、16 倍、32 倍),并限制在 `--outage-backoff-cap-seconds`(默认 3600s)及距离下一个 UTC 午夜的时间范围内。
```
# 完全禁用该功能
python3 wdgwars_api_tester.py --watch 1800 --outage-backoff-threshold 1.01
# 更激进:一旦 sweep 中有 10% 不佳就 back off,最多 sleep 4 小时
python3 wdgwars_api_tester.py --watch 1800 \
--outage-backoff-threshold 0.10 \
--outage-backoff-cap-seconds 14400
```
`DEAD`、`AUTH-REQUIRED`、`AUTH-REDIRECT` 和其他预期的非 OK 裁决不计入宕机比例 — 只有 `429` 和传输级别的 `ERROR` 才会计入。
## 通知渠道
`--watch` 模式支持三个独立的通知路径。可以同时使用一个、两个或全部三个 — 它们互不冲突。
| Flag | Use when |
|---|---|
| `--alert-telegram` | 你拥有 Telegram bot + chat。设置最简单。 |
| `--alert-webhook URL` | 你正在使用 Discord、Slack、n8n、PagerDuty 或任何接受 JSON POST 的服务。 |
| `--exec-on-change CMD` | 以上都不适合 — 邮件、短信、Lambda、写入数据库、输出到 logger 等。 |
任何一条路径失败都会向 stderr 记录警告,但绝不会使 watch 循环崩溃或阻塞其他路径。
### Telegram 自助寻呼
在 `--watch` 模式下,该工具可以在每次状态更改时直接发布到 Telegram chat。无需外部 broker、webhook 服务或告警基础设施 — 只需用标准库的 `urllib` 访问 Bot API 和一个 chat id 即可。
### 设置
1. 在 Telegram 上与 [@BotFather](https://t.me/BotFather) 对话并创建一个 bot。复制 token。
2. 将 bot 添加到你想要接收警报的 chat 中(私信、群组或频道)。
3. 在该 chat 中发送任何消息,然后 `GET https://api.telegram.org/bot/getUpdates` 并读取 `result[0].message.chat.id`(频道则读取 `result[0].channel_post.chat.id`)。
4. 导出这两个值并传入 `--alert-telegram`:
```
export TELEGRAM_BOT_TOKEN=123456:ABC-DEF...
export TELEGRAM_CHAT_ID=-1001234567890
python3 wdgwars_api_tester.py --watch 60 --alert-telegram
```
或者直接内联传入:`--telegram-bot-token --telegram-chat-id `。
### 消息格式
| Transition | Header |
|---|---|
| 恢复 (`* → HEALTHY`) | `✅ wdgwars API recovered` |
| 诊断故障(出现 `+SENTINEL-DIVERGED`) | `🔧 wdgwars-api-tester diagnostic broken` |
| 回归(其他任何恶化的情况) | `🚨 wdgwars API ` |
Body 包含 `prev_overall → curr_overall` 转换、单探测 delta(受 Telegram 4096 字符消息限制最多 30 行),以及裁决计数汇总。使用 HTML 解析模式,以便 `` / `
` 能够正确渲染。
### 通用 webhook (`--alert-webhook URL`)
在状态改变时向任何 HTTP endpoint POST JSON payload。该 payload 包含多个顶级键,因此同一个 URL 无需特定服务的标志即可适用于多个服务。从 v0.10.0 起,`text` + `content` 字段包含通俗易懂的英文描述,使得 Discord/Slack 频道读者无需上下文即可理解:
```
🚨 API status changed: all endpoints healthy → some endpoints down
What changed since the last check:
• team-me/valid: was healthy (HTTP 200), now timing out (>15s) or unreachable
• team-id/valid: was healthy (HTTP 200), now timing out (>15s) or unreachable
Current snapshot:
• 13 endpoints healthy
• 27 correctly rejecting unauthorized callers
• 2 timed out or unreachable
• 2 endpoints missing (404 sentinel match)
• 3 background API 404 sentinel (probe of /api/)
→ Non-upstream probe regressed. Investigate.
```
完整 payload:
```
{
"text": "🚨 API status changed: ... (the human-readable string above)",
"content": "",
"text_machine": "🚨 wdgwars-api-tester: HEALTHY → DEGRADED\n\n\n\nverdicts: DEAD=2, ERROR=2, OK=13, ...",
"title": "🚨 API status changed: ...",
"kind": "regression",
"overall": "DEGRADED",
"overall_human": "some endpoints down",
"prev_overall": "HEALTHY",
"prev_overall_human": "all endpoints healthy",
"deltas": ["wdgwars.pl team-me/valid OK/200 -> ERROR/-", "..."],
"deltas_human": ["team-me/valid: was healthy (HTTP 200), now timing out ..."],
"by_verdict": {"OK": 13, "AUTH-REQUIRED": 27, "ERROR": 2, "DEAD": 2},
"by_verdict_human": ["13 endpoints healthy", "27 correctly rejecting ...", "..."],
"action": "Non-upstream probe regressed. Investigate.",
"tool": "wdgwars-api-tester",
"version": "0.10.0"
}
```
- **Discord**读取 `content`。直接填入任何频道的 webhook URL 即可。
- **Slack incoming webhooks** 读取 `text`。同样直接填入即可。
- **n8n / Zapier / Make** 可以直接选取结构化字段。
- **PagerDuty Events v2** — 需配合 `--exec-on-change` 包装(它需要不同的信封格式)。
- **自定义 HTTP handlers** — 从结构化字段中读取它们需要的任何内容。
- **解析了旧版行话 `text` 的工具** — 改为读取 `text_machine`。格式与 v0.9.0 及更早版本相同。
### 每日摘要 (`--digest URL`)
Oneshot 模式:运行每个探测一次,并将每日摘要 POST 到 webhook。与在你期望接收摘要的本地时间(通常为 08:00)触发的 systemd timer 配合使用,这样 Discord/Slack 频道每天就能收到一条易于阅读的“昨晚发生了什么”的帖子。与 `--watch` 互斥。
```
# 将每次 --watch state change 追加到 state log 中
python3 wdgwars_api_tester.py --watch 1800 \
--alert-webhook "$DISCORD_LOUD_WEBHOOK" \
--state-log ~/wdgwars-api-tester/lab/state-log.jsonl
# 触发一次 digest(通常通过本地 08:00 的每日 systemd timer 执行)
python3 wdgwars_api_tester.py --digest "$DISCORD_LOUD_WEBHOOK" \
--state-log ~/wdgwars-api-tester/lab/state-log.jsonl
```
摘要读取的内容:
```
Morning report — 2026-06-04
API status right now: **all endpoints healthy**
50 probes ran.
• 14 endpoints healthy
• 27 correctly rejecting unauthorized callers
• 2 endpoints missing (404 sentinel match)
• 3 responding with 405 wrong-verb (endpoint healthy)
• 3 background API 404 sentinel (probe of /api/)
Last 24 hours: 3 state changes (2 loud, 1 suppressed as LOCOSP upstream flap)
• 1× HEALTHY → DEGRADED
• 1× DEGRADED → HEALTHY
Most-flapped probes:
• team-me/valid: 3 transitions
→ No action needed.
```
时间窗口可通过 `--digest-window-hours N` 配置(默认为 24)。
### 任意命令 (`--exec-on-change CMD`)
在状态改变时运行任何 shell 命令。以下环境变量将被导出到子进程中:
| Env var | Value |
|---|---|
| `WDGWARS_OVERALL` | 新的总体裁决,例如 `DEGRADED+LEAK` |
| `WDGWARS_PREV_OVERALL` | 之前的总体裁决 |
| `WDGWARS_KIND` | `recovery` / `regression` / `diagnostic-broken` |
| `WDGWARS_RECOVERY` | 如果转换为 HEALTHY 则为 `1`,否则为 `0` |
| `WDGWARS_DELTAS` | 换行符连接的单探测差异行 |
| `WDGWARS_VERDICTS` | JSON 编码的 `{verdict: count}` 字典 |
示例:
```
# 每次 transition 发送 Email
--exec-on-change 'echo "$WDGWARS_DELTAS" | mail -s "wdgwars: $WDGWARS_OVERALL" me@example.com'
# 仅在 regression 时告警(不包括 recovery,也不包括 diagnostic)
--exec-on-change '[ "$WDGWARS_KIND" = "regression" ] && /usr/local/bin/page-me.sh "$WDGWARS_OVERALL"'
# 转发到现有的内部 alerting service
--exec-on-change 'curl -X POST -H "Authorization: Bearer $MY_TOKEN" \
-d "{\"summary\":\"$WDGWARS_OVERALL\",\"verdicts\":$WDGWARS_VERDICTS}" \
https://internal.example.com/alert'
```
该命令以 `shell=True` 及 15 秒超时限制运行。非零退出码会记录警告,但不会导致 watch 循环崩溃。
## 将该工具适配为你自己的服务
单文件、MIT 许可证、仅依赖标准库 — 鼓励 fork。其结构设计旨在使这些修改变得简单:
- **探测不同的 API。** 编辑 `build_probes()` 以替换 endpoint、方法和预期状态。顶部的 `DEFAULT_HOSTS` / `ALL_HOSTS` 用于更改被探测的主机。
- **添加新探测。** 在 `build_probes()` 中追加 `Probe(...)` 条目。每个条目都会自动应用相同的认证变体矩阵和裁决标注。
- **添加新裁决。** 编辑 `annotate_verdicts()` 以添加分支,然后将该裁决添加到 `VERDICT_PRIORITY` 以确保表格排序正常工作,并在 `summary()` 中添加以便在相关时将其汇总到总体裁决中。
- **自定义哨兵机制。** `SENTINEL_PROBES` 和 `_canonical_sentinel()` 定义了法定逻辑。更改 `SENTINEL_PROBES` 以使用更多哨兵,或重写 `_canonical_sentinel()` 以使用不同的多数表决规则。
- **不同的通知格式。** 直接编辑 `_format_telegram_text()` 或 `_format_webhook_payload()`。两者都是纯函数,易于进行单元测试。
如果你发布了 fork,MIT 许可证意味着直接克隆和重命名是可以的 — 无需标明上游出处。
## 测试
包含两个测试套件,均仅使用标准库。
### 单元测试(离线,快速)
```
python3 -m unittest test_wdgwars_api_tester
```
包含 32 个测试,无需网络。涵盖裁决标注、法定哨兵逻辑、状态签名稳定性、摘要汇总、探测差异检测、Telegram 消息格式化以及 webhook payload 结构。运行耗时不到一秒钟。
### 集成测试(默认离线)
```
python3 integration_test.py # offline — fast, safe, default
python3 integration_test.py --live # also runs the live API check
INTEGRATION_LIVE=1 python3 integration_test.py # env var equivalent
```
21 个端到端场景。默认模式为**离线** — `integration_test.py` 会启动 `mock_wdgwars.py` 的本地实例(每个场景一个,使用随机端口)并将测试器指向它们。不会触及真实的 `wdgwars.pl`,因此该套件在每次 push 时运行都是安全的,不会给小型的社区托管 API 增加租户流量。
测试覆盖范围:
- `--version`, `--help`, 默认的 one-shot 模式, `--quiet`, `--json`, `--no-table`
- 无效 `--variants` / `--hosts` 的拒绝
- **针对特定场景的裁决断言** — 离线 mock 支持四种状态:
- `outage` → 测试器产生 `DEGRADED+LEAK`(若有有效 key 则为 `OUTAGE+LEAK`)
- `healthy` → 测试器产生 `HEALTHY`
- `partial` → 测试器产生 `HEALTHY+LEAK`(API 正常,但 stats endpoint 仍在泄露)
- `diverged` → 测试器产生带有 `+SENTINEL-DIVERGED` 后缀的结果
- `--baseline` 首次运行创建 + 第二次运行稳定性(同一场景下无差异)
- 所有三种通知防护机制(不带 `--watch` 而使用 `--alert-telegram` / `--alert-webhook` / `--exec-on-change` 会发出警告并禁用)
- 端到端的 webhook 分发 — 针对本地捕获服务器的 `_post_webhook`,payload 断言(Slack `text`,Discord `content`,结构化的 `kind`/`overall`/`prev_overall`)
- 端到端的 exec 分发 — 跨平台的 Python 环境捕获助手确认所有 `WDGWARS_*` 环境变量设置正确
- `--live` 选择性启用:针对真实 `wdgwars.pl` 的 schema 验证(每个已记录的探测均出现在 JSON 输出中,3 哨兵法定机制在未分歧的状态下产生 ≤2 个不同的哈希值)
离线运行耗时约 11 秒。`--live` 会针对真实的 `wdgwars.pl` 增加约 10-30 秒的一次实际探测耗时。退出码 0 = 全部通过。
### Mock server(独立使用)
`mock_wdgwars.py` 也可以作为独立的 HTTP server 运行,用于手动探索。这对于了解裁决形态或开发 fork 很有帮助:
```
python3 mock_wdgwars.py --scenario healthy --port 9999 &
python3 wdgwars_api_tester.py --hosts http://127.0.0.1:9999 --variants none,garbage
```
可用场景:`outage`、`healthy`、`partial`、`diverged`。
## 更新
仅依赖标准库,因此更新只需从 `main` 刷新单个 `.py` 文件:
```
./update.sh # Linux / Mac
update.bat # Windows (double-click)
```
或者手动更新:
```
curl -O https://raw.githubusercontent.com/HiroAlleyCat/wdgwars-api-tester/main/wdgwars_api_tester.py
```
## 相关
- [wigle-to-wdgwars](https://github.com/HiroAlleyCat/wigle-to-wdgwars) — WiFi/BLE CSV 上传器。
- [adsb-to-wdgwars (Muninn)](https://github.com/HiroAlleyCat/adsb-to-wdgwars) — ADS-B 上传器。
## 许可证
MIT。 标签:API监控, Python 3, 数据泄露防护, 状态检测, 网络探测, 运维工具