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, 数据泄露防护, 状态检测, 网络探测, 运维工具