diff --git a/CHANGELOG.md b/CHANGELOG.md index 9032fd4..2a0acf3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,8 +36,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (`ponte`) and the repository name are unchanged, so an install from a checkout behaves exactly as before. - **The coverage gate moved from 70% to 80%.** The `[[profiles]]`, notify and - doctor work pushed the suite past it (~83%), so the threshold in `pyproject.toml` + doctor work pushed the suite past it (~84%), so the threshold in `pyproject.toml` now matches the codebase instead of trailing it by ten points. +- **The tunnel target is part of the status now.** Each profile's status + carries the SSH `destination` it connects to (`ProfileStatus.destination`, + taken from the config rather than the status file), shown as the first row of + a `ponte status` table and included in `ponte status --json`. A table of + numbers is useless if you cannot tell which server the broken one is; the + dashboard labels every card with it. ### Added @@ -105,6 +111,25 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 is non-zero when anything failed, so it is usable from a script. - `[windows] pythonw_exe` — pin the windowless interpreter for the Scheduled Task, for installs where `pythonw.exe` does not sit next to `python.exe`. +- **`ponte serve` — a local HTTP surface: dashboard, health probe and + Prometheus metrics.** `/` renders a self-contained dashboard (inline CSS, no + CDN, no JavaScript) with a card per tunnel: health, session age, availability, + forwarded-port state, the last disconnect and its reason, and the event feed. + `/healthz` is the endpoint to point a monitor at — it returns `503` when the + daemon is down *or* a tunnel is broken, and `200` (with `"status": + "starting"`) until the first health check completes, so a restart does not + page anyone. `/metrics` speaks the Prometheus text format (session age, + cumulative up/down time, availability, reconnect and port-listening state) + and deliberately always answers `200`, because a scrape failure would hide + *why* a tunnel went down — a graph's whole job. `/status.json` is exactly the + `ponte status --json` payload, so the page, the probe and the metrics can + never disagree with the CLI. Everything is read-only and re-read per request + (`Cache-Control: no-store`), and the whole thing is stdlib `http.server` — no + new dependency. Binds `127.0.0.1` by default; `[serve] host`/`port`/ + `token`/`refresh` configure it, and a non-loopback bind **without** a token is + refused at config load and by `ponte serve` alike (the dashboard names your + servers, users and ports), with clients then passing `?token=` or + `Authorization: Bearer`. ### Planned diff --git a/README.md b/README.md index c7f9aa6..8d6fd74 100644 --- a/README.md +++ b/README.md @@ -56,6 +56,11 @@ - 🩺 **`ponte doctor`** — one command that checks the config, the key file and its permissions, SSH reachability, listening ports, auto-start status and the notify channel, each row ending in a concrete fix instead of a black box. +- 📊 **A dashboard and a metrics endpoint** — `ponte serve` puts the same status + on HTTP: `/` is a self-contained dashboard (no CDN, no JavaScript), `/healthz` + answers `503` when a tunnel is actually broken, `/metrics` speaks Prometheus + and `/status.json` is exactly `ponte status --json`. Loopback-only by default; + exposing it needs an explicit token. - 🖥️ **Cross-platform** — resolves `ssh` automatically, per-platform runtime paths, and portable remote-port probing (`socket` → `ss`/`lsof`/`netstat`). @@ -90,6 +95,7 @@ import package stay `ponte`; a checkout installs the same way (`pipx install .`) | `check [--profile NAME]` | verify tunnel ports are listening (`-R` on the server, `-L`/`-D` locally) | | `doctor [--offline] [--timeout S]` | one-shot checkup of config, key, connectivity, ports, auto-start and notifications, each row with a fix | | `notify-test [--profile NAME]` | send a test alert through the configured ntfy / webhook channels | +| `serve [--host H] [--port P] [--token T] [--open]` | local HTTP dashboard, `/healthz` probe, Prometheus `/metrics`, `/status.json` snapshot | | `install` / `uninstall` | register / remove the OS auto-start service | | `config` | print the effective configuration, its source file and any warnings | @@ -97,6 +103,61 @@ Global options (before the command): `--config/-c PATH` pin a config file, `--version/-V` print the version. Unknown/typo'd config keys are reported by `ponte config` instead of being silently ignored. +## 📊 Web dashboard & monitoring + +`ponte watch` is for the machine you are sitting at; `ponte serve` is for +everything else — a browser, a phone on the same host, Uptime Kuma, Prometheus. + +```bash +ponte serve # http://127.0.0.1:8787/ (loopback only by default) +ponte serve --open # ...and open it in your browser +``` + +| Endpoint | What it answers | +|----------|-----------------| +| `/` | the dashboard: per-tunnel health, session age, availability, port state, last disconnect with its reason, event feed | +| `/healthz` | `200` while the tunnels work, `503` as soon as one is broken — the endpoint to point a monitor at | +| `/metrics` | Prometheus text exposition: session age, cumulative up/down time, availability, reconnects, port-listening state | +| `/status.json` | exactly the payload of `ponte status --json` | + +**Why `/healthz` and `/metrics` disagree on purpose.** `/healthz` fails, so a +monitor can alert; `/metrics` always answers `200` and reports state as numbers, +because a scrape failure would hide *why* a tunnel went down — which is exactly +what a graph exists to show. And `/healthz` reports `starting` (with `200`) until +the first health check completes, so restarting the daemon does not page you. + +**Security.** The dashboard names your servers, users and forwarded ports — it +is a map of your infrastructure, not a status line. So `ponte serve` binds +`127.0.0.1` and nothing else. Binding a LAN or public address is possible, but +only together with a token; ponte *refuses* the combination of "exposed" and +"no token" instead of warning about it: + +```toml +[serve] +host = "0.0.0.0" # opt in, deliberately +port = 8787 +token = "a-long-random-string" # required for any non-loopback host +refresh = 5 # dashboard auto-refresh, seconds +# ipv6 hosts are fine too: host = "::1" +``` + +Clients then pass `?token=...` (handy for scrapers) or `Authorization: Bearer +...`. All four endpoints are read-only, re-read the daemon status per request and +send `Cache-Control: no-store`, so a page can never show a stale "healthy" for a +tunnel that has since died. + +```yaml +# prometheus.yml +scrape_configs: + - job_name: ponte + static_configs: + - targets: ["127.0.0.1:8787"] + # with a token: metrics_path: /metrics?token=a-long-random-string +``` + +Alert on `ponte_profile_port_listening == 0` for the signal that matters most: +a live process whose forwarded port is gone is the classic silent failure. + ## 🛠️ Cross-platform service management | Platform | Mechanism | Generated artifact | @@ -128,6 +189,8 @@ ponte (local daemon, Python) - `health.py` — periodic liveness + remote-port checks - `notify.py` — ntfy / webhook alerts on repeated failures - `doctor.py` — one-shot diagnostics used by `ponte doctor` +- `serve.py` — read-only HTTP surface (dashboard / health probe / metrics) over + the same payload `ponte status --json` emits - `config.py` — TOML load/validate (built-in `tomllib` on 3.11+) ## ⚙️ Configuration @@ -186,6 +249,9 @@ Sections: | Connection rejected after key change | delete `known_hosts`, reconnect (`StrictHostKeyChecking=accept-new` default) | | Process alive but remote port down | cloud security-group inbound rules; check server with `ss -tlnp` / `lsof -nP -iTCP -sTCP:LISTEN` — the daemon now force-reconnects a "zombie" tunnel after 3 consecutive failed checks | | Console window flashes at logon, or while stopping | the Scheduled Task must run `pythonw.exe` — check `[windows] pythonw_exe`; `ponte stop` also force-kills through a hidden `taskkill` | +| `ponte serve` exits with "cannot bind" / port busy | another process holds the port — `ponte serve --port 8788`; the refused non-loopback bind is a *token* problem, and the message says so | +| `/healthz` returns `401` | a `[serve].token` is set: pass `?token=...` or `Authorization: Bearer ...` | +| `/healthz` returns `503` while the tunnel looks fine | it reports the *tunnel*, not the process: read `unhealthy` / `errors` in the body, then `ponte check` | | Logs | `ponte logs -n 100 --follow` | ## 🧪 Development & testing @@ -251,6 +317,10 @@ again. See [CONTRIBUTING.md](CONTRIBUTING.md). 让你在真出事**之前**就验证通道可用。默认关闭：不开启就绝不会外发任何数据。 - 🩺 **`ponte doctor`** — 一条命令逐项体检：配置、密钥及其权限、SSH 连通性、 监听端口、开机自启状态、通知通道，每行都给出具体修法而不是留个黑箱。 +- 📊 **看板与指标接口** — `ponte serve` 把同一份状态摆到 HTTP 上：`/` 是自包含的 + 看板（不依赖 CDN、不用 JavaScript），`/healthz` 在隧道真的断时回 `503`， + `/metrics` 说 Prometheus 格式，`/status.json` 就是 `ponte status --json`。 + 默认只监听本机；要对外必须先给令牌。 - 🖥️ **跨平台** — 自动查找 `ssh`、按平台落盘运行时文件、可移植的远程端口探测 （`socket` → `ss`/`lsof`/`netstat`）。 @@ -284,6 +354,7 @@ ponte install # 注册开机自启 + 崩溃重启 | `check [--profile NAME]` | 检查隧道端口（`-R` 在服务器上，`-L`/`-D` 在本机） | | `doctor [--offline] [--timeout S]` | 一键体检配置、密钥、连通性、端口、自启与通知，每项给出修法 | | `notify-test [--profile NAME]` | 通过已配置的 ntfy / webhook 通道发一条测试通知 | +| `serve [--host H] [--port P] [--token T] [--open]` | 本地 HTTP 看板、`/healthz` 探活、Prometheus `/metrics`、`/status.json` 快照 | | `install` / `uninstall` | 注册 / 移除开机自启服务 | | `config` | 打印生效配置、来源文件与配置告警 | @@ -291,6 +362,58 @@ ponte install # 注册开机自启 + 崩溃重启 `--version/-V` 打印版本。拼错/未知的配置项会由 `ponte config` 报出来， 不再被静默忽略。 +## 📊 网页看板与监控接入 + +`ponte watch` 给坐在机器前的你看，`ponte serve` 给其它一切：浏览器、手机 +（同机）、Uptime Kuma、Prometheus。 + +```bash +ponte serve # http://127.0.0.1:8787/（默认只监听本机） +ponte serve --open # 顺手在浏览器里打开 +``` + +| 接口 | 回答什么问题 | +|------|--------------| +| `/` | 看板：逐条隧道的健康、当前会话时长、在线率、端口状态、上次断线原因与事件流 | +| `/healthz` | 隧道正常时 `200`，任一条断开立即 `503` —— 监控就探这个 | +| `/metrics` | Prometheus 文本格式：会话时长、累计在线/离线、在线率、重连次数、端口监听状态 | +| `/status.json` | 与 `ponte status --json` 完全一致的载荷 | + +**为什么 `/healthz` 与 `/metrics` 故意不一致。** `/healthz` 会失败，监控才能 +报警；`/metrics` 永远回 `200`，把状态当数字报出来——因为采挂掉会盖住 +“它为何挂了”，而那正是画图的目的。另外首次健康检查完成前，`/healthz` +报的是 `starting`（`200`），所以重启守护进程不会造成误报。 + +**安全模型。** 看板会列出你的服务器地址、登录用户与转发端口——这是一张 +内网拓扑图，不是一行状态。所以 `ponte serve` 只绑 `127.0.0.1`。绑到局域网或 +公网是可以的，但**必须**同时给令牌：ponte 对“对外 + 无令牌”的组合是直接 +拒绝，而不是警告一句了事。 + +```toml +[serve] +host = "0.0.0.0" # 显式选择对外 +port = 8787 +token = "一个足够长的随机串" # 非回环地址必需 +refresh = 5 # 看板自动刷新秒数 +# 也支持 IPv6：host = "::1" +``` + +客户端用 `?token=...`（脚本/采集器方便）或 `Authorization: Bearer ...`。 +四个接口全是只读、每次请求都重新读取守护进程状态，并带 +`Cache-Control: no-store`——所以页面不会拿旧的“健康”去骗一个已经挂了的隧道。 + +```yaml +# prometheus.yml +scrape_configs: + - job_name: ponte + static_configs: + - targets: ["127.0.0.1:8787"] + # 带令牌时：metrics_path: /metrics?token=一个足够长的随机串 +``` + +最值得拿来报警的一条是 `ponte_profile_port_listening == 0`：进程活着、 +转发端口却没了，正是那种悄无声息的典型故障。 + ## 🛠️ 跨平台服务管理 | 平台 | 机制 | 生成物 | @@ -325,6 +448,8 @@ ponte（本地守护进程，Python） - `health.py` — 周期存活 + 远程端口检查 - `notify.py` — 连续失败时的 ntfy / webhook 告警 - `doctor.py` — `ponte doctor` 使用的体检项 +- `serve.py` — 只读 HTTP 接口（看板 / 探活 / 指标），渲染的就是 + `ponte status --json` 那份载荷 - `config.py` — TOML 加载/校验（3.11+ 内置 `tomllib`） ## ⚙️ 配置 @@ -363,6 +488,8 @@ ponte（本地守护进程，Python） `cooldown`（同一 profile 两条告警之间的最小秒数），以及通道： `ntfy_topic`（可选 `ntfy_server` / `ntfy_token`）和/或 `webhook_url` （以 JSON 形式收到告警） +- `[serve]` — 本地看板：`host`（默认 `127.0.0.1`）、`port`（默认 `8787`）、 + `token`（绑定非回环地址时必填，否则拒绝启动）、`refresh`（看板刷新秒数） - `[service]` — 服务名、自启、POSIX 强杀等待 - `[windows]` — 仅 Windows 使用（`task_name`、`ssh_exe`、`pythonw_exe`、 `run_as`）。`run_as` 默认 `user`（登录后以你本人身份运行、能读 `~/.ssh`）或 @@ -378,6 +505,9 @@ ponte（本地守护进程，Python） | 换 key 后连接被拒 | 删除 `known_hosts` 重连（默认 `StrictHostKeyChecking=accept-new`） | | 进程活着但远程端口不通 | 云安全组入方向规则；服务器上 `ss -tlnp` / `lsof -nP -iTCP -sTCP:LISTEN` 确认监听 —— 守护进程已支持假死检测：连续 3 次检查失败自动强制重连 | | 登录时（或 `stop` 时）闪出黑色控制台窗口 | 计划任务必须跑 `pythonw.exe`——检查 `[windows] pythonw_exe`；`ponte stop` 的强杀也已隐藏控制台 | +| `ponte serve` 报绑定失败 / 端口占用 | 换端口：`ponte serve --port 8788`；若报的是非回环地址，那是**令牌**问题，报错里写了 | +| `/healthz` 返回 `401` | 配了 `[serve].token`：带上 `?token=...` 或 `Authorization: Bearer ...` | +| 隧道看着正常，`/healthz` 却回 `503` | 它报的是**隧道**不是进程：看响应体里的 `unhealthy` / `errors`，再用 `ponte check` 复核 | | 排查日志 | `ponte logs -n 100 --follow` | ## 🧪 开发与测试 diff --git a/ponte/config.example.toml b/ponte/config.example.toml index bee1b62..e701686 100644 --- a/ponte/config.example.toml +++ b/ponte/config.example.toml @@ -143,6 +143,16 @@ cooldown = 900 # webhook_url = "https://example.com/hooks/ponte" # 两个通道可以同时配。配好后用 `ponte notify-test` 验证真能收到。 +[serve] +# 本地 HTTP 看板 / 指标（ponte serve）。默认只监听本机：看板会列出服务器 +# 地址、登录用户与转发端口，等于一张内网拓扑图，因此默认不对外。 +host = "127.0.0.1" +port = 8787 +# 想让局域网或反向代理访问就必须设一个令牌，否则 ponte serve 直接拒绝启动。 +# 客户端用 ?token=... 或 Authorization: Bearer ... 带上它。 +# token = "一个足够长的随机串" +refresh = 5 # 看板自动刷新间隔（秒）；接口本身从不缓存 + [service] # 开机自启服务身份（Linux systemd 单元名 / macOS launchd label 后缀） name = "ponte" diff --git a/ponte/config.py b/ponte/config.py index cbd1950..8753fb6 100644 --- a/ponte/config.py +++ b/ponte/config.py @@ -52,6 +52,9 @@ "HealthConfig", "WindowsConfig", "ServiceConfig", + "ServeConfig", + "ensure_bindable", + "is_loopback_host", "get_config", "load_config", "set_config_path", @@ -357,6 +360,36 @@ def channels(self) -> tuple[str, ...]: return tuple(found) +@dataclass(frozen=True) +class ServeConfig: + """The optional local HTTP surface (``ponte serve``). + + ``host`` defaults to loopback because every page under it names your + servers, users and forwarded ports — that is a map of your infrastructure, + not a status line. Binding anywhere else is allowed but *demands* a + ``token``: :func:`ensure_bindable` refuses rather than warns, for the same + reason the daemon refuses to install a task that would flash a console. + + ``refresh`` is the dashboard's self-refresh interval in seconds. The JSON, + health and metrics endpoints are never cached, so this only affects how + often a browser redraws itself. + """ + + host: str = "127.0.0.1" + port: int = 8787 + token: str = "" + refresh: int = 5 + + @property + def loopback(self) -> bool: + """``True`` when this bind address is reachable from this machine only.""" + return is_loopback_host(self.host) + + def check_bind(self) -> None: + """Raise :class:`ConfigValidationError` for an exposing configuration.""" + ensure_bindable(self.host, self.token) + + @dataclass(frozen=True) class WindowsConfig: """Platform specific knobs used only on Windows. @@ -579,6 +612,7 @@ class TunnelConfig: notify: NotifyConfig = field(default_factory=NotifyConfig) windows: WindowsConfig = field(default_factory=WindowsConfig) service: ServiceConfig = field(default_factory=ServiceConfig) + serve: ServeConfig = field(default_factory=ServeConfig) source_path: str = "" """Absolute path of the TOML file this configuration was loaded from.""" warnings: tuple[str, ...] = () @@ -695,6 +729,7 @@ def _parse_config(data: Mapping[str, Any], config_path: str) -> TunnelConfig: notify = _parse_notify(data.get("notify", {}), warnings) windows = _parse_windows(data.get("windows", {}), warnings) service = _parse_service(data.get("service", {}), warnings) + serve = _parse_serve(data.get("serve", {}), warnings) cfg = TunnelConfig( profiles=profiles, @@ -704,6 +739,7 @@ def _parse_config(data: Mapping[str, Any], config_path: str) -> TunnelConfig: notify=notify, windows=windows, service=service, + serve=serve, source_path=config_path, warnings=tuple(warnings), ) @@ -726,6 +762,7 @@ def _parse_config(data: Mapping[str, Any], config_path: str) -> TunnelConfig: "notify", "windows", "service", + "serve", } ) _KNOWN_PROFILE = frozenset({"name", "ssh", "tunnels"}) @@ -757,6 +794,7 @@ def _parse_config(data: Mapping[str, Any], config_path: str) -> TunnelConfig: ) _KNOWN_WINDOWS = frozenset({"task_name", "ssh_exe", "pythonw_exe", "run_as"}) _KNOWN_SERVICE = frozenset({"name", "autostart", "kill_timeout"}) +_KNOWN_SERVE = frozenset({"host", "port", "token", "refresh"}) def _warn_unknown_keys( @@ -1051,6 +1089,26 @@ def _parse_notify(section: Any, warnings: list[str] | None = None) -> NotifyConf return notify +def _parse_serve(section: Any, warnings: list[str] | None = None) -> ServeConfig: + if not section: + return ServeConfig() + _expect_table(section, "serve") + _warn_unknown_keys(section, _KNOWN_SERVE, "serve", warnings) + dft = ServeConfig() + serve = ServeConfig( + host=_optional_str(section, "host", default=dft.host), + port=_optional_int( + section, "port", default=dft.port, minimum=1, maximum=65535, where="serve" + ), + token=_optional_str(section, "token", default=""), + refresh=_optional_int( + section, "refresh", default=dft.refresh, minimum=1, where="serve" + ), + ) + serve.check_bind() + return serve + + def _parse_windows(section: Any, warnings: list[str] | None = None) -> WindowsConfig: if not section: return WindowsConfig() @@ -1088,6 +1146,33 @@ def _parse_service(section: Any, warnings: list[str] | None = None) -> ServiceCo ) +def is_loopback_host(host: str) -> bool: + """Return ``True`` when *host* binds to this machine only. + + Deliberately conservative: anything not obviously loopback is treated as + exposed. That includes ``""``, which ``http.server`` reads as *all + interfaces*, so an odd spelling cannot sneak past the token requirement in + :func:`ensure_bindable`. + """ + name = host.strip().strip("[]").lower() + return name in ("localhost", "::1") or name.startswith("127.") + + +def ensure_bindable(host: str, token: str) -> None: + """Refuse to expose the status surface to the network without a token. + + Raises: + ConfigValidationError: when *host* is not loopback and *token* is empty. + """ + if is_loopback_host(host) or token: + return + raise ConfigValidationError( + f"serve.host = {host!r} 会把看板暴露到网络上，必须同时设置 serve.token" + "（或命令行 --token）：看板会列出服务器地址、登录用户与转发端口，" + "没有令牌等于把内网拓扑摊开给同网段的任何人。" + ) + + def _validate(cfg: TunnelConfig) -> None: """Cross-field validation that runs after every section is parsed.""" if not cfg.profiles: diff --git a/ponte/daemon.py b/ponte/daemon.py index 4d2fd28..a5043d3 100644 --- a/ponte/daemon.py +++ b/ponte/daemon.py @@ -185,6 +185,11 @@ class ProfileStatus: """ name: str + #: ``user@host:port`` this profile connects to, taken from the + #: configuration (not the status file). It is what makes a multi-profile + #: ``status --json`` / dashboard row self-describing: a table of numbers is + #: useless if you cannot tell which server is the broken one. + destination: str | None = None #: ``None`` until the first health check of this profile reports in. healthy: bool | None = None process_alive: bool | None = None @@ -289,7 +294,9 @@ def get_profile(self, name: str) -> ProfileStatus | None: return None -def _profile_status(name: str, section: dict) -> ProfileStatus: +def _profile_status( + name: str, section: dict, *, destination: str | None = None +) -> ProfileStatus: """Build a :class:`ProfileStatus` from one status-file section. Tolerant on purpose: the file is written by whichever daemon version is @@ -325,6 +332,7 @@ def _ports(key: str) -> dict[int, bool]: raw_alive = section.get("process_alive") return ProfileStatus( name=name, + destination=destination, healthy=raw_healthy if isinstance(raw_healthy, bool) else None, process_alive=raw_alive if isinstance(raw_alive, bool) else None, remote_ports=_ports("remote_ports"), @@ -1129,13 +1137,19 @@ def status(self) -> DaemonStatus: # daemon that still supervises it is stopped. names = list(self.profile_names) names += [name for name in sections if name not in names] + destinations = { + profile.name: profile.destination for profile in self.config.profiles + } return DaemonStatus( running=True, pid=pid, started_at=started, uptime_seconds=max(0.0, uptime), profiles=[ - _profile_status(name, sections.get(name, {})) for name in names + _profile_status( + name, sections.get(name, {}), destination=destinations.get(name) + ) + for name in names ], ) diff --git a/ponte/main.py b/ponte/main.py index 63d8eb1..eaa3a74 100644 --- a/ponte/main.py +++ b/ponte/main.py @@ -10,10 +10,12 @@ from __future__ import annotations +import dataclasses import json import os import sys import time +import webbrowser from pathlib import Path from typing import TYPE_CHECKING, NoReturn @@ -25,9 +27,16 @@ from rich.table import Table from ponte import __version__ -from ponte.config import ConfigError, get_config, init_config, set_config_path +from ponte.config import ( + ConfigError, + ServeConfig, + get_config, + init_config, + set_config_path, +) from ponte.daemon import _format_duration from ponte.doctor import FAIL, OK, SKIP, WARN, counts, run_checks +from ponte.serve import create_server, serve_url if TYPE_CHECKING: # pragma: no cover - import cycle guard, runtime import is lazy from ponte.daemon import TunnelDaemon @@ -283,6 +292,9 @@ def _add_daemon_rows(table: Table, s) -> None: # noqa: ANN001 - DaemonStatus cy def _add_profile_rows(table: Table, profile) -> None: # noqa: ANN001 - cycle guard """Append one profile's health, statistics and port states to *table*.""" + # 目标放在第一行：多条隧道时，最先要说清的是“这张表是哪个服务器”。 + if profile.destination: + table.add_row("目标", escape(profile.destination)) table.add_row("健康状态", _markup_health(profile.healthy, profile.health_error)) # 会话时长是区分“守护进程活了多久”与“隧道活了多久”的那一列。 @@ -323,6 +335,7 @@ def _round1(value: float | None) -> float | None: def _profile_payload(profile) -> dict: # noqa: ANN001 - ProfileStatus cycle guard """Machine-readable snapshot of one profile (the ``--json`` contract).""" return { + "destination": profile.destination, "healthy": profile.healthy, "process_alive": profile.process_alive, "health_error": profile.health_error, @@ -556,6 +569,96 @@ def watch( _fail(str(exc)) +# --------------------------------------------------------------------------- +# serve（本地 HTTP 看板 / 指标） +# --------------------------------------------------------------------------- + + +def _serve_config( + base: ServeConfig, + *, + host: str | None, + port: int | None, + token: str | None, + refresh: int | None, +) -> ServeConfig: + """Apply ``ponte serve``'s command-line overrides to the ``[serve]`` section.""" + return dataclasses.replace( + base, + host=base.host if host is None else host, + port=base.port if port is None else port, + token=base.token if token is None else token, + refresh=base.refresh if refresh is None else refresh, + ) + + +@app.command() +def serve( + host: str | None = typer.Option( + None, "--host", help="监听地址（默认取 [serve].host，也就是只监听本机）" + ), + port: int | None = typer.Option( + None, "--port", min=1, max=65535, help="监听端口（默认取 [serve].port）" + ), + token: str | None = typer.Option( + None, + "--token", + help="访问令牌；绑定非回环地址时必须提供（写在命令行上会进 shell 历史，" + "长期使用建议写进 [serve].token）", + ), + refresh: int | None = typer.Option( + None, "--refresh", min=1, help="看板自动刷新间隔（秒）" + ), + open_browser: bool = typer.Option(False, "--open", help="启动后在浏览器里打开看板"), +) -> None: + """启动本地 HTTP 服务：看板 / 、探活 /healthz、指标 /metrics、快照 /status.json。""" + try: + daemon = _daemon() + effective = _serve_config( + daemon.config.serve, host=host, port=port, token=token, refresh=refresh + ) + # 与配置文件走同一条校验：绑定非回环地址却没有令牌时直接拒绝， + # 不提供“先跑起来再说”的选项。 + effective.check_bind() + except typer.Exit: + raise + except Exception as exc: + _fail(str(exc)) + + try: + server = create_server( + lambda: _status_payload(daemon.status()), + host=effective.host, + port=effective.port, + token=effective.token, + refresh=effective.refresh, + ) + except OSError as exc: + _fail( + f"无法监听 {effective.host}:{effective.port}（{exc}）；" + "换一个端口：ponte serve --port 8788" + ) + + url = serve_url(effective.host, effective.port) + console.print(f"[green]ponte 看板已启动：{url}[/green]") + console.print( + f"[dim]指标 {url}metrics · 探活 {url}healthz · 快照 {url}status.json[/dim]" + ) + if not effective.loopback: + console.print( + "[yellow]警告：已绑定非回环地址，同网段里拿到令牌的人都能看到你的服务器、" + "用户与端口；不要暴露到公网[/yellow]" + ) + if open_browser: + webbrowser.open(url) + try: + server.serve_forever(poll_interval=0.5) + except KeyboardInterrupt: + console.print("\n[yellow]已停止看板[/yellow]") + finally: + server.server_close() + + # --------------------------------------------------------------------------- # test / check # --------------------------------------------------------------------------- @@ -781,6 +884,14 @@ def config() -> None: f"on_consecutive_failures={notify.on_consecutive_failures}, " f"cooldown={notify.cooldown}s", ) + # 令牌只报“有没有”，绝不回显：ponte config 的输出经常被粘进 issue。 + serve = cfg.serve + table.add_row( + "serve", + f"host={serve.host}, port={serve.port}, " + f"token={'已设置' if serve.token else '（无）'}, " + f"refresh={serve.refresh}s", + ) table.add_row("pid_file", cfg.daemon.pid_file or "（默认）") table.add_row("log_file", cfg.daemon.log_file or "（默认）") table.add_row("ssh_exe", cfg.windows.ssh_exe or "ssh（PATH）") diff --git a/ponte/serve.py b/ponte/serve.py new file mode 100644 index 0000000..5c0e24f --- /dev/null +++ b/ponte/serve.py @@ -0,0 +1,985 @@ +"""A local HTTP surface for ponte: dashboard, health probe and metrics. + +``ponte watch`` answers "is my tunnel up?" for a human sitting at the machine. +This module answers it for everything else: + +* ``/`` a self-contained HTML dashboard — no CDN, no JavaScript, no + external assets, so it renders from ``curl``, a phone + browser on the same host, or an air-gapped box; +* ``/healthz`` a probe for uptime monitors, whose status code reports + whether the *tunnel* works, not whether a process exists; +* ``/metrics`` Prometheus text exposition, for a Prometheus / Grafana / + VictoriaMetrics scrape; +* ``/status.json`` byte-for-byte the payload of ``ponte status --json``. + +Three rules shape the implementation: + +* **Loopback by default, and a token to leave it.** Everything under ``/`` + names your servers, users and forwarded ports. Binding a non-loopback address + without a token is refused up front by :func:`ponte.config.ensure_bindable` + rather than served with a warning nobody reads. +* **Standard library only.** ``http.server``, not a web framework: ponte exists + to be dropped on a machine that has nothing but SSH and Python, and a + dependency would cost more than this feature is worth. +* **Read-only, and always fresh.** Every request re-reads the daemon status, so + a page can never show a cached "healthy" for a tunnel that has since died. + Nothing here can start, stop or reconfigure anything — the worst a leaked + token buys is a read of your port numbers. + +The one thing this module deliberately does *not* do is invent numbers: the +dashboard, the JSON, the health probe and the metrics are all rendered from the +same payload, so the web page can never disagree with ``ponte status``. +""" + +from __future__ import annotations + +import contextlib +import hmac +import html +import json +import logging +import socket +import time +from collections.abc import Callable, Iterable, Mapping +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from string import Template +from typing import Any, cast +from urllib.parse import parse_qs, urlparse + +from ponte import __version__ +from ponte.config import ensure_bindable +from ponte.daemon import _format_duration + +__all__ = [ + "PonteHTTPServer", + "create_server", + "dashboard_html", + "health_response", + "render_metrics", + "serve_url", +] + +logger = logging.getLogger(__name__) + +#: How many recent retry-loop events a dashboard card shows. +_FEED_LIMIT = 8 + +#: The routes this server answers. Anything else is a 404 — an explicit list +#: rather than a fallback, so a typo never silently serves something. +_ROUTES = ("/", "/healthz", "/metrics", "/status.json") + +#: Event type → (glyph, tone). Mirrors ``ponte watch`` so the two views teach +#: the same visual language. +_EVENT_GLYPHS: dict[str, tuple[str, str]] = { + "connecting": ("→", "dim"), + "connected": ("●", "ok"), + "disconnected": ("●", "bad"), + "retrying": ("↻", "warn"), + "max_retries_reached": ("✗", "bad"), +} + + +# --------------------------------------------------------------------------- +# Payload accessors +# +# The payload is the ``ponte status --json`` contract. These helpers are +# tolerant on purpose: the JSON is written by whichever daemon version is +# installed, so a missing or unexpected field degrades to "unknown" instead of +# turning the dashboard into a stack trace while the user is trying to read +# their status. +# --------------------------------------------------------------------------- + + +def _profiles(payload: Mapping[str, Any]) -> dict[str, Mapping[str, Any]]: + """Return the ``name → section`` map of a status payload.""" + raw = payload.get("profiles") + if not isinstance(raw, Mapping): + return {} + return { + str(name): section + for name, section in raw.items() + if isinstance(section, Mapping) + } + + +def _as_float(value: Any, default: float | None = None) -> float | None: + """Coerce a JSON number, returning *default* for anything else.""" + if isinstance(value, bool) or not isinstance(value, (int, float)): + return default + return float(value) + + +def _session_age(section: Mapping[str, Any], now: float) -> float | None: + """Seconds the current session has been up, ``None`` when disconnected.""" + started = _as_float(section.get("current_session_at")) + if started is None: + return None + return max(0.0, now - started) + + +# --------------------------------------------------------------------------- +# /healthz +# --------------------------------------------------------------------------- + + +def health_response(payload: Mapping[str, Any]) -> tuple[int, dict[str, Any]]: + """Build ``(status_code, body)`` for ``/healthz``. + + The code reports whether the tunnel works, which is the whole reason to + have this endpoint instead of pinging the PID file: + + * ``503 down`` — the daemon is not running: nothing is being forwarded. + * ``503 degraded`` — running, but at least one profile is unhealthy. + * ``200 ok`` — running, and every profile is healthy. + * ``200 starting`` — running, but no health check has reported yet. A fresh + ``ponte start`` waits up to ``check_interval`` (60 s by default) for its + first answer; calling that "down" would fire a false alert on every + restart, and "no data yet" is not evidence of failure. + """ + if not payload.get("running"): + return 503, {"status": "down", "reason": "ponte daemon is not running"} + + profiles = _profiles(payload) + if not profiles: + return 200, {"status": "starting", "reason": "no profile has reported yet"} + + unhealthy = sorted( + name for name, section in profiles.items() if section.get("healthy") is False + ) + if unhealthy: + errors = { + name: profiles[name].get("health_error") + for name in unhealthy + if profiles[name].get("health_error") + } + return 503, { + "status": "degraded", + "profiles": len(profiles), + "unhealthy": unhealthy, + "errors": errors, + } + + if all(section.get("healthy") is True for section in profiles.values()): + return 200, {"status": "ok", "profiles": len(profiles)} + return 200, { + "status": "starting", + "profiles": len(profiles), + "reason": "no health check has completed yet", + } + + +# --------------------------------------------------------------------------- +# /metrics +# --------------------------------------------------------------------------- + + +def _escape_label(value: Any) -> str: + """Escape a Prometheus label value (backslash, newline, double quote).""" + return ( + str(value).replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"') + ) + + +def _metric_value(value: Any) -> str | None: + """Format *value* as a Prometheus sample value, ``None`` when unknown. + + A missing number is *omitted* rather than exported as NaN: a gap in a graph + says "no data", while a NaN line invites the reader to guess. + """ + if isinstance(value, bool): + return "1" if value else "0" + if isinstance(value, int): + return str(value) + if isinstance(value, float): + return f"{value:.3f}" + return None + + +def _series(name: str, labels: Mapping[str, Any], value: Any) -> str | None: + """One sample line, or ``None`` when the value is unknown.""" + formatted = _metric_value(value) + if formatted is None: + return None + if not labels: + return f"{name} {formatted}" + rendered = ",".join( + f'{key}="{_escape_label(val)}"' for key, val in labels.items() + ) + return f"{name}{{{rendered}}} {formatted}" + + +class _Families: + """Accumulates metric families so samples come out grouped. + + The text exposition format requires every sample of a metric to be one + uninterrupted group, which is why this is not simply a list of lines. A + family whose samples are all unknown is dropped entirely — an empty + ``# HELP`` header with no samples is noise in every graph it reaches. + """ + + def __init__(self) -> None: + self._order: list[str] = [] + self._families: dict[str, list[Any]] = {} + + def add( + self, + name: str, + kind: str, + help_text: str, + samples: Iterable[str | None], + ) -> None: + """Register *samples* under the family *name* (``gauge``/``counter``).""" + kept = [sample for sample in samples if sample] + if not kept: + return + if name not in self._families: + self._order.append(name) + # HELP text stays English: it is read by Prometheus/Grafana + # tooling, not by the operator's terminal. + self._families[name] = [kind, help_text, []] + entry = self._families[name] + entry[2].extend(kept) + + def render(self) -> str: + """The complete exposition text, ending with a newline.""" + lines: list[str] = [] + for name in self._order: + kind, help_text, samples = self._families[name] + lines.append(f"# HELP {name} {help_text}") + lines.append(f"# TYPE {name} {kind}") + lines.extend(samples) + return "\n".join(lines) + "\n" + + +def render_metrics(payload: Mapping[str, Any], *, now: float | None = None) -> str: + """Render a status payload as Prometheus text exposition (``0.0.4``). + + Deliberately *not* the same signal as ``/healthz``: that one fails when the + tunnel is broken, so a monitor can alert. This one always answers ``200`` + and reports the state as numbers, because a scrape failure would hide + *why* the tunnel went down — exactly the question a graph exists to answer. + """ + moment = time.time() if now is None else now + families = _Families() + + running = bool(payload.get("running")) + families.add( + "ponte_up", + "gauge", + "1 when the ponte daemon is running.", + [_series("ponte_up", {}, running)], + ) + families.add( + "ponte_build_info", + "gauge", + "ponte version, always 1 (join on this to graph deploys).", + [_series("ponte_build_info", {"version": __version__}, True)], + ) + families.add( + "ponte_daemon_uptime_seconds", + "gauge", + "Seconds since the ponte daemon process started.", + [_series("ponte_daemon_uptime_seconds", {}, payload.get("uptime_seconds"))], + ) + if not running: + return families.render() + + profiles = _profiles(payload) + families.add( + "ponte_profiles_configured", + "gauge", + "Profiles the daemon supervises.", + [_series("ponte_profiles_configured", {}, len(profiles))], + ) + families.add( + "ponte_profiles_unhealthy", + "gauge", + "Profiles that are currently unhealthy.", + [ + _series( + "ponte_profiles_unhealthy", + {}, + sum( + 1 + for section in profiles.values() + if section.get("healthy") is False + ), + ) + ], + ) + # Identity lives in one info metric instead of being repeated as a label on + # every series: the samples stay keyed by profile (what a legend wants), and + # a scrape does not re-encode the same destination a dozen times. + families.add( + "ponte_profile_info", + "gauge", + "Static identity of a profile: the SSH destination it connects to.", + [ + _series( + "ponte_profile_info", + {"profile": name, "destination": section.get("destination") or ""}, + True, + ) + for name, section in profiles.items() + ], + ) + + def add( + name: str, + kind: str, + help_text: str, + getter: Callable[[Mapping[str, Any]], Any], + ) -> None: + """Register one numeric family, one sample per profile.""" + families.add( + name, + kind, + help_text, + [ + _series(name, {"profile": profile}, getter(section)) + for profile, section in profiles.items() + ], + ) + + add( + "ponte_profile_healthy", + "gauge", + "1 when every health check of the profile passes; absent until the first check.", + lambda section: section.get("healthy"), + ) + add( + "ponte_profile_process_alive", + "gauge", + "1 while the profile's SSH process is alive.", + lambda section: section.get("process_alive"), + ) + add( + "ponte_profile_session_uptime_seconds", + "gauge", + "Age of the current SSH session; absent while the tunnel is down.", + lambda section: _session_age(section, moment), + ) + add( + "ponte_profile_availability_ratio", + "gauge", + "Completed uptime / observed time for the profile (0..1), the same ratio ponte status prints.", + lambda section: section.get("availability"), + ) + add( + "ponte_profile_connect_attempts_total", + "counter", + "SSH launch attempts (cumulative; survives daemon restarts).", + lambda section: section.get("connect_attempts_total"), + ) + add( + "ponte_profile_sessions_total", + "counter", + "SSH sessions actually established.", + lambda section: section.get("sessions_total"), + ) + add( + "ponte_profile_reconnects_total", + "counter", + "Scheduled reconnects after a drop.", + lambda section: section.get("reconnects_total"), + ) + add( + "ponte_profile_tunnel_uptime_seconds", + "counter", + "Cumulative seconds the tunnel has been up.", + lambda section: section.get("tunnel_uptime_seconds"), + ) + add( + "ponte_profile_tunnel_downtime_seconds", + "counter", + "Cumulative seconds the tunnel has been down.", + lambda section: section.get("tunnel_downtime_seconds"), + ) + add( + "ponte_profile_last_disconnect_timestamp_seconds", + "gauge", + "Unix time of the most recent disconnect.", + lambda section: section.get("last_disconnect_at"), + ) + add( + "ponte_profile_last_notification_timestamp_seconds", + "gauge", + "Unix time the most recent failure alert was delivered.", + lambda section: section.get("last_notification_at"), + ) + + # The strongest signal of the lot: whether the port is *actually* forwarded. + # A healthy-looking process with a dead port is the classic silent failure. + port_samples: list[str | None] = [] + for name, section in profiles.items(): + for key, kind in (("remote_ports", "remote"), ("local_ports", "local")): + ports = section.get(key) + if not isinstance(ports, Mapping): + continue + for port, listening in ports.items(): + port_samples.append( + _series( + "ponte_profile_port_listening", + {"profile": name, "kind": kind, "port": port}, + listening, + ) + ) + families.add( + "ponte_profile_port_listening", + "gauge", + "1 when a forwarded port is listening (remote: on the server, local: on this host).", + port_samples, + ) + return families.render() + + +# --------------------------------------------------------------------------- +# The dashboard +# --------------------------------------------------------------------------- + +#: The dashboard shell. A :class:`string.Template` (``$name`` placeholders) +#: rather than ``str.format`` because the stylesheet is full of braces; +#: ``substitute`` is used on purpose so a stray placeholder fails loudly in +#: tests instead of rendering as literal text. +_PAGE = Template( + """ + + + + + + +ponte · $title + + + + +

ponte v$version

$summary
+
$cards
+ + + +""" +) + + +def _esc(value: Any) -> str: + """Escape a value for HTML text or attribute context. + + Everything that reaches the page goes through this: profile names come from + the config, and disconnect reasons come from SSH's stderr — neither is + trusted input. + """ + return html.escape(str(value), quote=True) + + +def _pill(text: str, tone: str) -> str: + """A coloured status pill (``tone`` is ``ok``/``bad``/``unknown``).""" + return f'{_esc(text)}' + + +def _health_pill(section: Mapping[str, Any]) -> str: + """The health pill of one profile.""" + healthy = section.get("healthy") + if healthy is True: + return _pill("健康", "ok") + if healthy is False: + return _pill("异常", "bad") + return _pill("未知", "unknown") + + +def _kv(label: str, value: str) -> str: + """One table row. *value* must already be escaped or be trusted markup.""" + return f"{_esc(label)}{value}" + + +def _row(label: str, value: Any) -> str: + """One table row from a plain (untrusted) value.""" + return _kv(label, _esc(value)) + + +def _port_rows(section: Mapping[str, Any]) -> str: + """Rows for the forwarded ``-R``/``-L``/``-D`` ports of one profile.""" + rows: list[str] = [] + for key, label in (("remote_ports", "远程端口"), ("local_ports", "本地端口")): + ports = section.get(key) + if not isinstance(ports, Mapping) or not ports: + continue + for port, listening in sorted(ports.items(), key=_port_order): + rows.append( + _kv( + f"{label} {port}", + _pill("监听中", "ok") if listening else _pill("未监听", "bad"), + ) + ) + return "".join(rows) + + +def _port_order(item: tuple[Any, Any]) -> int: + """Sort key putting ports in numeric order (the JSON keys are strings).""" + try: + return int(item[0]) + except (TypeError, ValueError): # pragma: no cover - defensive + return 0 + + +def _feed(section: Mapping[str, Any]) -> str: + """The recent-event feed of one profile, newest first.""" + events = section.get("recent_events") + if not isinstance(events, list) or not events: + return '

暂无事件

' + lines: list[str] = [] + for event in reversed(events[-_FEED_LIMIT:]): + if not isinstance(event, Mapping): # pragma: no cover - defensive + continue + etype = str(event.get("type", "?")) + glyph, tone = _EVENT_GLYPHS.get(etype, ("·", "dim")) + at = _as_float(event.get("at"), 0.0) or 0.0 + stamp = time.strftime("%H:%M:%S", time.localtime(at)) + detail = etype + if event.get("reason"): + detail = f"{etype}: {event['reason']}" + elif event.get("attempt"): + delay = _as_float(event.get("delay"), 0.0) or 0.0 + detail = f"{etype}: 第 {event['attempt']} 次，{delay:.1f}s 后重试" + lines.append( + f'
{_esc(stamp)}' + f'{_esc(glyph)}{_esc(detail)}
' + ) + return "".join(lines) + + +def _profile_card(name: str, section: Mapping[str, Any], *, now: float) -> str: + """One profile's card: identity, statistics, ports and event feed.""" + healthy = section.get("healthy") + tone = "ok" if healthy is True else ("bad" if healthy is False else "unknown") + rows: list[str] = [] + + if section.get("destination"): + rows.append(_row("目标", section["destination"])) + if section.get("process_alive") is not None: + rows.append( + _kv( + "SSH 进程", + _pill("运行中", "ok") + if section.get("process_alive") + else _pill("已退出", "bad"), + ) + ) + + started = _as_float(section.get("current_session_at")) + if started is None: + rows.append(_kv("当前会话", _pill("已断开", "bad"))) + else: + rows.append(_row("当前会话", _format_duration(now - started))) + + availability = _as_float(section.get("availability")) + if availability is not None: + rows.append(_row("在线率", f"{availability * 100:.1f}%")) + + if section.get("sessions_total") is not None: + rows.append( + _row( + "会话统计", + f"会话 {section.get('sessions_total')} 次 · " + f"重连 {section.get('reconnects_total')} 次 · " + f"启动尝试 {section.get('connect_attempts_total')} 次", + ) + ) + + up_total = _as_float(section.get("tunnel_uptime_seconds")) + if up_total is not None: + down_total = _as_float(section.get("tunnel_downtime_seconds"), 0.0) or 0.0 + rows.append( + _row( + "累计在线 / 离线", + f"{_format_duration(up_total)} / {_format_duration(down_total)}", + ) + ) + + rows.append(_port_rows(section)) + + if section.get("last_disconnect_reason"): + detail = str(section["last_disconnect_reason"]) + at = _as_float(section.get("last_disconnect_at")) + if at is not None: + detail += ( + f"（{_format_duration(now - at)}前，" + f"{time.strftime('%m-%d %H:%M:%S', time.localtime(at))}）" + ) + rows.append(_row("上次断线", detail)) + + notified = _as_float(section.get("last_notification_at")) + if notified is not None: + rows.append(_row("上次通知", f"{_format_duration(now - notified)}前")) + + if section.get("health_error"): + rows.append(_row("检查错误", section["health_error"])) + if section.get("error"): + rows.append(_row("循环错误", section["error"])) + + return ( + f'
' + f"

{_esc(name)} {_health_pill(section)}

" + f'{"".join(rows)}
' + f'

最近事件

{_feed(section)}
' + f"
" + ) + + +def _summary(payload: Mapping[str, Any], profiles: Mapping[str, Any]) -> str: + """The header line: overall state, pid, daemon uptime and profile count.""" + if not payload.get("running"): + return _pill("守护进程未运行", "bad") + '可执行 ponte start 启动' + unhealthy = [ + name for name, section in profiles.items() if section.get("healthy") is False + ] + if unhealthy: + overall = _pill(f"{len(unhealthy)}/{len(profiles)} 条隧道异常", "bad") + elif profiles and all( + section.get("healthy") is True for section in profiles.values() + ): + overall = _pill("全部健康", "ok") + else: + overall = _pill("等待首次检查", "unknown") + pieces = [overall] + if payload.get("pid") is not None: + pieces.append(f'pid {_esc(payload["pid"])}') + uptime = _as_float(payload.get("uptime_seconds")) + if uptime is not None: + pieces.append(f'守护进程运行 {_esc(_format_duration(uptime))}') + pieces.append(f'{len(profiles)} 条隧道') + return "".join(pieces) + + +def dashboard_html( + payload: Mapping[str, Any], + *, + refresh: int = 5, + now: float | None = None, +) -> str: + """Render the whole dashboard page from a ``ponte status`` payload.""" + moment = time.time() if now is None else now + profiles = _profiles(payload) + if not payload.get("running"): + cards: list[str] = [ + '

守护进程未运行

' + '

先执行 ponte start 启动隧道，' + "本页面会在下一轮自动刷新。

" + ] + else: + cards = [ + _profile_card(name, section, now=moment) + for name, section in profiles.items() + ] + if not cards: + cards = [ + '

尚无隧道上报

' + '

守护进程已启动，等待第一次健康检查。

' + ] + return _PAGE.substitute( + refresh=max(1, int(refresh)), + version=_esc(__version__), + title="隧道看板", + summary=_summary(payload, profiles), + cards="".join(cards), + ) + + +# --------------------------------------------------------------------------- +# The HTTP server +# --------------------------------------------------------------------------- + + +def serve_url(host: str, port: int, path: str = "/") -> str: + """The URL to open for a bind *host*/*port* (``""`` means IPv4 loopback).""" + if not host: + return f"http://127.0.0.1:{port}{path}" + if ":" in host and not host.startswith("["): + return f"http://[{host}]:{port}{path}" + return f"http://{host}:{port}{path}" + + +def _address_family(host: str) -> int: + """Pick the socket family for *host* (IPv6 literals contain a colon).""" + return socket.AF_INET6 if ":" in host else socket.AF_INET + + +class PonteHTTPServer(ThreadingHTTPServer): + """The server ``ponte serve`` runs, carrying its own status provider. + + Subclassing rather than closing over the provider in the handler keeps the + request handlers plain methods — and lets a test start the server on port + ``0`` and read back the port the OS assigned. + """ + + daemon_threads = True + allow_reuse_address = True + + def __init__( + self, + address: tuple[str, int], + provider: Callable[[], dict[str, Any]], + *, + token: str = "", + refresh: int = 5, + ) -> None: + # Set before ``super().__init__``: TCPServer creates its socket there, + # so an IPv6 loopback (``::1``) has to pick its family first. + self.address_family = _address_family(address[0]) + self.provider = provider + self.token = token + self.refresh = refresh + super().__init__(address, PonteRequestHandler) + + +class PonteRequestHandler(BaseHTTPRequestHandler): + """Serves the four read-only endpoints; one instance per connection.""" + + server_version = f"ponte/{__version__}" + sys_version = "" + # HTTP/1.1 keeps the connection alive for a scraper's next request; every + # response sends an explicit Content-Length, which is what that requires. + protocol_version = "HTTP/1.1" + + @property + def _state(self) -> PonteHTTPServer: + """The owning server, typed so the handler can read its settings.""" + return cast(PonteHTTPServer, self.server) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + """Route the stdlib's stderr chatter into the ponte logger.""" + logger.debug("%s %s", self.address_string(), format % args) + + # -- methods ----------------------------------------------------------- + + def do_GET(self) -> None: # noqa: N802 - stdlib naming + self._respond(head=False) + + def do_HEAD(self) -> None: # noqa: N802 + self._respond(head=True) + + def do_POST(self) -> None: # noqa: N802 + self._method_not_allowed() + + def do_PUT(self) -> None: # noqa: N802 + self._method_not_allowed() + + def do_PATCH(self) -> None: # noqa: N802 + self._method_not_allowed() + + def do_DELETE(self) -> None: # noqa: N802 + self._method_not_allowed() + + def _method_not_allowed(self) -> None: + """Everything here is read-only, and says so instead of pretending.""" + self._send_json( + 405, + {"error": "read-only endpoint; use GET"}, + extra={"Allow": "GET, HEAD"}, + ) + + # -- routing ----------------------------------------------------------- + + def _respond(self, *, head: bool) -> None: + """Route one request: 404 → 401 → the endpoint.""" + parsed = urlparse(self.path) + path = parsed.path.rstrip("/") or "/" + if path not in _ROUTES: + self._send_json( + 404, + {"error": "not found", "endpoints": list(_ROUTES)}, + head=head, + ) + return + if not self._authorized(parse_qs(parsed.query)): + self._send_json( + 401, + {"error": "unauthorized: missing or invalid token"}, + head=head, + extra={"WWW-Authenticate": 'Bearer realm="ponte"'}, + ) + return + + try: + payload = self._state.provider() + except Exception as exc: # noqa: BLE001 - answer, never drop the client + logger.exception("serve: status provider failed") + self._send_json( + 500, {"error": f"status unavailable: {type(exc).__name__}"}, head=head + ) + return + + if path in ("/", "/index.html"): + self._send_text( + 200, + dashboard_html(payload, refresh=self._state.refresh), + "text/html; charset=utf-8", + head=head, + ) + elif path == "/healthz": + code, body = health_response(payload) + self._send_json(code, body, head=head) + elif path == "/metrics": + self._send_text( + 200, + render_metrics(payload), + "text/plain; version=0.0.4; charset=utf-8", + head=head, + ) + else: # "/status.json" + self._send_json(200, dict(payload), head=head) + + def _authorized(self, query: Mapping[str, list[str]]) -> bool: + """Check the token in constant time, from the header or ``?token=``. + + The query string exists because Prometheus scrapes a URL and per-target + ``Authorization`` headers are awkward to configure; the header is what a + hand-written client would use. + """ + expected = self._state.token + if not expected: + return True + supplied = "" + header = self.headers.get("Authorization", "") + if header.lower().startswith("bearer "): + supplied = header[7:].strip() + if not supplied: + supplied = (query.get("token") or [""])[0] + return hmac.compare_digest( + supplied.encode("utf-8"), expected.encode("utf-8") + ) + + # -- responses --------------------------------------------------------- + + def _send( + self, + code: int, + body: bytes, + content_type: str, + *, + head: bool, + extra: Mapping[str, str] | None = None, + ) -> None: + """Send one complete response (never cached).""" + self.send_response(code) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(len(body))) + # Status data must never be cached: a browser or proxy replaying a + # "healthy" page for a tunnel that has since died is exactly the + # failure this endpoint exists to prevent. + self.send_header("Cache-Control", "no-store") + for key, value in (extra or {}).items(): + self.send_header(key, value) + self.end_headers() + if head: + return + with contextlib.suppress(BrokenPipeError, ConnectionResetError): + self.wfile.write(body) + + def _send_json( + self, + code: int, + payload: Mapping[str, Any], + *, + head: bool = False, + extra: Mapping[str, str] | None = None, + ) -> None: + """Send a JSON response (compact: this is machine food).""" + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + self._send( + code, + body, + "application/json; charset=utf-8", + head=head, + extra=extra, + ) + + def _send_text( + self, + code: int, + text: str, + content_type: str, + *, + head: bool, + ) -> None: + """Send a text response (HTML or metrics).""" + self._send(code, text.encode("utf-8"), content_type, head=head) + + +def create_server( + provider: Callable[[], dict[str, Any]], + *, + host: str = "127.0.0.1", + port: int = 8787, + token: str = "", + refresh: int = 5, +) -> PonteHTTPServer: + """Build (but do not start) the dashboard server. + + The caller owns the lifecycle; ``ponte serve`` runs ``serve_forever`` on it, + a test calls ``serve_forever`` on a thread and shuts it down again. + + Raises: + ConfigValidationError: when *host* is not loopback and *token* is empty. + OSError: when the address is unusable or the port is already taken. + """ + ensure_bindable(host, token) + return PonteHTTPServer((host, port), provider, token=token, refresh=refresh) diff --git a/tests/test_config.py b/tests/test_config.py index a35ada3..528948f 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -12,7 +12,9 @@ ConfigParseError, ConfigValidationError, TunnelConfig, + ensure_bindable, get_config, + is_loopback_host, load_config, ) @@ -929,3 +931,132 @@ def test_profile_invalid_identity_file_is_named(tmp_path) -> None: ) + _profile_entry(tmp_path, "db") with pytest.raises(ConfigValidationError, match="web"): load_config(_write_toml(tmp_path, body)) + + +# --------------------------------------------------------------------------- +# [serve] —— 本地 HTTP 看板 +# --------------------------------------------------------------------------- + + +def _serve_config_file(tmp_path, section: str) -> str: + """一份最小可用配置，后面接上任意的 ``[serve]`` 段文本。""" + return _tunnels_config( + tmp_path, + """ +[[tunnels]] +remote_port = 23334 +local_host = "localhost" +local_port = 2222 + +""" + + section, + ) + + +def test_serve_defaults_to_loopback(tmp_path) -> None: + """没有 [serve] 段时默认只监听本机：看板不需要用户做任何决定就应该是安全的。""" + serve = load_config(_minimal(tmp_path)).serve + assert serve.host == "127.0.0.1" + assert serve.port == 8787 + assert serve.token == "" + assert serve.loopback is True + + +def test_serve_section_is_parsed(tmp_path) -> None: + path = _serve_config_file( + tmp_path, + """ +[serve] +host = "192.168.1.5" +port = 9100 +token = "s3cret" +refresh = 15 +""", + ) + serve = load_config(path).serve + assert serve.host == "192.168.1.5" + assert serve.port == 9100 + assert serve.token == "s3cret" + assert serve.refresh == 15 + assert serve.loopback is False + + +def test_serve_non_loopback_without_token_is_rejected(tmp_path) -> None: + """拒绝而不是警告：看板会列出服务器、用户与端口。""" + path = _serve_config_file( + tmp_path, + """ +[serve] +host = "0.0.0.0" +""", + ) + with pytest.raises(ConfigValidationError, match="token"): + load_config(path) + + +def test_serve_allows_an_explicit_non_loopback_bind_with_token(tmp_path) -> None: + """带上令牌就允许对外，把选择权交给用户而不是替他决定。""" + path = _serve_config_file( + tmp_path, + """ +[serve] +host = "0.0.0.0" +token = "s3cret" +""", + ) + assert load_config(path).serve.host == "0.0.0.0" + + +def test_serve_port_must_be_in_range(tmp_path) -> None: + path = _serve_config_file( + tmp_path, + """ +[serve] +port = 70000 +""", + ) + with pytest.raises(ConfigValidationError, match="serve.port"): + load_config(path) + + +def test_serve_unknown_key_warns(tmp_path) -> None: + """拼错的键要报出来，而不是默默用默认值。""" + path = _serve_config_file( + tmp_path, + """ +[serve] +hsot = "127.0.0.1" +""", + ) + cfg = load_config(path) + assert any("serve.hsot" in warning for warning in cfg.warnings) + + +@pytest.mark.parametrize( + ("host", "expected"), + [ + ("127.0.0.1", True), + ("127.1.2.3", True), + ("localhost", True), + ("LOCALHOST", True), + ("[::1]", True), + ("", False), + ("0.0.0.0", False), + ("::", False), + ("192.168.1.5", False), + ("example.com", False), + ], +) +def test_is_loopback_host(host, expected) -> None: + """白名单判定：不认识的写法一律当成“对外暴露”。 + + ``""`` 特别重要：``http.server`` 把空地址当成 *所有网卡*，所以它绝不是 + 回环地址，不能因为“看起来是空的”就放行。 + """ + assert is_loopback_host(host) is expected + + +def test_ensure_bindable_treats_an_empty_host_as_exposed() -> None: + with pytest.raises(ConfigValidationError, match="token"): + ensure_bindable("", "") + ensure_bindable("", "s3cret") diff --git a/tests/test_daemon.py b/tests/test_daemon.py index fa3ee88..af119ab 100644 --- a/tests/test_daemon.py +++ b/tests/test_daemon.py @@ -1072,3 +1072,36 @@ def test_service_tool_calls_go_through_run_tool(monkeypatch, tmp_path) -> None: assert args == ["systemctl", "--user", "daemon-reload"] assert kwargs["creationflags"] == creation_flags() assert kwargs["capture_output"] is True and kwargs["text"] is True + + +def test_status_names_the_target_of_each_profile(tmp_path) -> None: + """每条隧道都要能说出自己连的是哪台服务器。 + + 目标来自配置而不是状态文件（状态文件里就没有这个信息），所以它不能 + 因为 daemon 刚重启、还没上报而消失——否则看板上会出现一栏“健康的 + 未知服务器”，而多隧道时那正是最需要弄清楚的一件事。 + """ + cfg = _two_profile_cfg(tmp_path) + d = TunnelDaemon(cfg) + _live_pid(tmp_path) + d._store.begin(["web", "db"], started_at=time.time()) + + s = d.status() + + assert [profile.destination for profile in s.profiles] == [ + profile.destination for profile in cfg.profiles + ] + # 字面断言（而不是子串包含）：既钉死格式，也更严格。 + assert s.profiles[0].destination == "testuser@web.example.com" + + +def test_status_leaves_the_target_unknown_for_a_dropped_profile(tmp_path) -> None: + """配置里已删掉的 profile 还会显示（daemon 仍管着它），但没有目标可报。""" + d = TunnelDaemon(_cfg(tmp_path)) + _live_pid(tmp_path) + d._store.begin(["default", "ghost"], started_at=time.time()) + + ghost = d.status().get_profile("ghost") + + assert ghost is not None + assert ghost.destination is None diff --git a/tests/test_main.py b/tests/test_main.py index 22e3461..909e5ce 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -13,6 +13,7 @@ HealthConfig, Profile, RetryConfig, + ServeConfig, SSHConfig, SSHOptions, Tunnel, @@ -554,3 +555,197 @@ def _interrupt(_seconds: float) -> None: result = CliRunner().invoke(app, ["logs", "--follow"]) assert result.exit_code == 0 assert "已停止跟随" in result.output + + +# --------------------------------------------------------------------------- +# ponte serve（本地 HTTP 看板） +# --------------------------------------------------------------------------- + + +class _FakeServeDaemon: + """``ponte serve`` 只用到 ``config.serve`` 与 ``status()``，假对象给这两个。""" + + def __init__(self, serve: ServeConfig | None = None) -> None: + self.config = dataclasses.replace( + _cfg(), serve=serve if serve is not None else ServeConfig() + ) + + def status(self) -> DaemonStatus: + return _status( + ProfileStatus( + name="default", + destination="testuser@example.com:22", + healthy=True, + ), + running=True, + pid=4242, + uptime_seconds=10.0, + ) + + +class _FakeServer: + """记录 CLI 怎么启动/收尾服务器，不碰真 socket。""" + + def __init__(self, **kwargs) -> None: + self.kwargs = kwargs + self.forever = False + self.closed = False + + def serve_forever(self, poll_interval: float = 0.5) -> None: + self.forever = True + raise KeyboardInterrupt + + def server_close(self) -> None: + self.closed = True + + +def _record_server(record: dict): + """返回一个可注入的 ``create_server``，把参数与 provider 记进 *record*。""" + + def _create(provider, **kwargs) -> _FakeServer: + record["provider"] = provider + server = _FakeServer(**kwargs) + record["server"] = server + return server + + return _create + + +def test_serve_is_registered_in_help(monkeypatch) -> None: + monkeypatch.setattr("ponte.main.get_config", lambda: _cfg()) + result = CliRunner().invoke(app, ["--help"]) + assert result.exit_code == 0 + assert "serve" in result.output + + +def test_serve_command_serves_and_stops_cleanly(monkeypatch) -> None: + monkeypatch.setattr("ponte.main._daemon", lambda: _FakeServeDaemon(ServeConfig(port=9100))) + record: dict = {} + monkeypatch.setattr("ponte.main.create_server", _record_server(record)) + + result = CliRunner().invoke(app, ["serve"]) + + assert result.exit_code == 0 + assert "http://127.0.0.1:9100/" in result.output + assert record["server"].closed is True + assert record["server"].kwargs["port"] == 9100 + # provider 必须现读现取：看板的“新鲜度”全靠它，缓存一次就失去意义。 + payload = record["provider"]() + assert payload["profiles"]["default"]["healthy"] is True + + +def test_serve_refuses_to_expose_without_a_token(monkeypatch) -> None: + """绑定非回环地址而没有令牌：直接拒绝，不提供“先跑起来再说”。""" + monkeypatch.setattr("ponte.main._daemon", lambda: _FakeServeDaemon()) + monkeypatch.setattr("ponte.main.create_server", _record_server({})) + + result = CliRunner().invoke(app, ["serve", "--host", "0.0.0.0"]) + + assert result.exit_code == 1 + assert "token" in result.output + + +def test_serve_command_line_overrides_win_and_warn(monkeypatch) -> None: + """命令行参数覆盖 [serve]，并就把看板摆到局域网的后果给出警告。""" + monkeypatch.setattr("ponte.main._daemon", lambda: _FakeServeDaemon(ServeConfig(port=8787))) + record: dict = {} + monkeypatch.setattr("ponte.main.create_server", _record_server(record)) + + result = CliRunner().invoke( + app, + [ + "serve", + "--host", "0.0.0.0", + "--port", "9100", + "--token", "s3cret", + "--refresh", "30", + ], + ) + + assert result.exit_code == 0 + assert record["server"].kwargs == { + "host": "0.0.0.0", + "port": 9100, + "token": "s3cret", + "refresh": 30, + } + assert "警告" in result.output + + +def test_serve_open_launches_the_browser(monkeypatch) -> None: + monkeypatch.setattr("ponte.main._daemon", lambda: _FakeServeDaemon(ServeConfig(port=9100))) + monkeypatch.setattr("ponte.main.create_server", _record_server({})) + opened: list[str] = [] + monkeypatch.setattr("ponte.main.webbrowser.open", opened.append) + + result = CliRunner().invoke(app, ["serve", "--open"]) + + assert result.exit_code == 0 + assert opened == ["http://127.0.0.1:9100/"] + + +def test_serve_config_applies_only_given_overrides() -> None: + from ponte.main import _serve_config + + base = ServeConfig(host="127.0.0.1", port=9100, token="t", refresh=7) + assert _serve_config(base, host=None, port=None, token=None, refresh=None) == base + changed = _serve_config(base, host="0.0.0.0", port=1, token="x", refresh=2) + assert changed == ServeConfig(host="0.0.0.0", port=1, token="x", refresh=2) + + +def test_config_command_reports_serve_without_leaking_the_token(monkeypatch) -> None: + """ponte config 的输出经常被粘进 issue：令牌只报“有没有”。""" + cfg = dataclasses.replace( + _cfg(), + serve=ServeConfig(host="0.0.0.0", port=9100, token="s3cret", refresh=15), + ) + monkeypatch.setattr("ponte.main.get_config", lambda: cfg) + result = CliRunner().invoke(app, ["config"]) + assert result.exit_code == 0 + assert "9100" in result.output + assert "已设置" in result.output + assert "s3cret" not in result.output + + +def test_status_shows_the_target_destination(monkeypatch) -> None: + """多隧道时，最先要说清的是“这张表是哪个服务器”。""" + s = _status( + ProfileStatus( + name="default", destination="testuser@example.com:22", healthy=True + ), + running=True, + pid=1, + uptime_seconds=10.0, + ) + + class _Daemon: + def status(self) -> DaemonStatus: + return s + + monkeypatch.setattr("ponte.main._daemon", lambda: _Daemon()) + result = CliRunner().invoke(app, ["status"]) + assert result.exit_code == 0 + assert "testuser@example.com:22" in result.output + + +def test_status_json_includes_destination(monkeypatch) -> None: + """--json 里也要有目标：监控脚本靠它区分是哪条隧道。""" + import json as _json + + s = _status( + ProfileStatus( + name="default", destination="testuser@example.com:22", healthy=True + ), + running=True, + pid=7, + ) + + class _Daemon: + def status(self) -> DaemonStatus: + return s + + monkeypatch.setattr("ponte.main._daemon", lambda: _Daemon()) + result = CliRunner().invoke(app, ["status", "--json"]) + assert result.exit_code == 0 + payload = _json.loads(result.output) + assert payload["profiles"]["default"]["destination"] == "testuser@example.com:22" diff --git a/tests/test_serve.py b/tests/test_serve.py new file mode 100644 index 0000000..ddf9cac --- /dev/null +++ b/tests/test_serve.py @@ -0,0 +1,534 @@ +"""Tests for ``ponte serve`` — the HTTP dashboard, health probe and metrics. + +The renderers are pure functions over a ``ponte status --json`` payload, so most +of this file needs no sockets. The end-to-end section starts a *real* server on +an OS-assigned port and speaks real HTTP to it, because the parts worth being +sure about — routing, status codes, the token gate, the absence of caching — +only exist once bytes go over a socket. +""" + +from __future__ import annotations + +import contextlib +import json +import re +import threading +import urllib.error +import urllib.request +from collections.abc import Iterator + +import pytest + +from ponte import __version__ +from ponte.config import ConfigValidationError, ensure_bindable +from ponte.serve import ( + create_server, + dashboard_html, + health_response, + render_metrics, + serve_url, +) + +_NOW = 1_700_000_000.0 + + +def _profile(**overrides) -> dict: + """One profile section, shaped exactly like ``ponte status --json``.""" + section = { + "destination": "deploy@example.com:22", + "healthy": True, + "process_alive": True, + "health_error": None, + "error": None, + "remote_ports": {"23334": True}, + "local_ports": {"1080": False}, + "connect_attempts_total": 4, + "sessions_total": 3, + "reconnects_total": 1, + "tunnel_uptime_seconds": 3600.0, + "tunnel_downtime_seconds": 60.0, + "availability": 0.9, + "current_session_at": _NOW - 120.0, + "last_disconnect_at": _NOW - 600.0, + "last_disconnect_reason": "ssh exited with code 255", + "last_notification_at": None, + "recent_events": [ + {"at": _NOW - 700.0, "type": "disconnected", "reason": "ssh exited with code 255"}, + {"at": _NOW - 690.0, "type": "retrying", "attempt": 1, "delay": 5.0}, + {"at": _NOW - 120.0, "type": "connected"}, + ], + } + section.update(overrides) + return section + + +def _payload(**overrides) -> dict: + """A whole-daemon payload.""" + base = { + "running": True, + "pid": 4242, + "started_at": _NOW - 3600.0, + "uptime_seconds": 3600.0, + "healthy": True, + "profiles": {"web": _profile()}, + } + base.update(overrides) + return base + + +def _parse_families(text: str) -> dict[str, dict]: + """Parse exposition text, asserting the format's grouping rules holds. + + Returns ``{name: {"type": str, "samples": [line, ...]}}``. The assertions + inside are the real test: a metric's samples must form one uninterrupted + group, every family needs exactly one ``# HELP``/``# TYPE``, and a sample + may never appear before its own header. + """ + families: dict[str, dict] = {} + current: str | None = None + for line in text.splitlines(): + if line.startswith("# HELP "): + name, _, help_text = line[len("# HELP ") :].partition(" ") + assert name not in families, f"duplicate HELP for {name}" + assert help_text.strip(), f"empty HELP for {name}" + families[name] = {"type": None, "samples": [], "help": help_text} + current = name + elif line.startswith("# TYPE "): + name, _, kind = line[len("# TYPE ") :].partition(" ") + assert families[name]["type"] is None, f"duplicate TYPE for {name}" + families[name]["type"] = kind + current = name + elif line.strip(): + name = line.split("{")[0].split(" ")[0] + assert name == current, f"sample {name} is not grouped under {current}" + families[current]["samples"].append(line) + for name, family in families.items(): + assert family["type"] in ("gauge", "counter"), name + assert family["samples"], f"{name} has no samples" + return families + + +#: ``key="value"`` with backslash escapes honoured, i.e. what a Prometheus +#: parser sees. Values keep their escapes; :func:`_unescape` undoes them. +_LABEL_PATTERN = re.compile(r'([A-Za-z_][A-Za-z0-9_]*)="((?:[^"\\]|\\.)*)"') + + +def _labels(line: str) -> dict[str, str]: + """Parse the label set of a sample line (``{}`` when it has none).""" + return { + match.group(1): match.group(2) + for match in _LABEL_PATTERN.finditer(line) + } + + +def _unescape(value: str) -> str: + """Undo Prometheus label escaping (``\\n``, ``\\"``, ``\\\\``).""" + out: list[str] = [] + index = 0 + while index < len(value): + char = value[index] + if char == "\\" and index + 1 < len(value): + nxt = value[index + 1] + out.append({"n": "\n", '"': '"', "\\": "\\"}.get(nxt, nxt)) + index += 2 + continue + out.append(char) + index += 1 + return "".join(out) + + +def _sample(families: dict[str, dict], name: str, **labels: str) -> str | None: + """Return the sample of *name* whose labels include *labels*.""" + for line in families[name]["samples"]: + parsed = _labels(line) + if all(parsed.get(key) == value for key, value in labels.items()): + return line + return None + + +# --------------------------------------------------------------------------- +# /healthz +# --------------------------------------------------------------------------- + + +def test_healthz_down_when_daemon_is_not_running() -> None: + code, body = health_response({"running": False}) + assert code == 503 + assert body["status"] == "down" + + +def test_healthz_ok_when_every_profile_is_healthy() -> None: + code, body = health_response(_payload()) + assert code == 200 + assert body == {"status": "ok", "profiles": 1} + + +def test_healthz_degraded_names_the_broken_profile() -> None: + """A 503 has to say *which* tunnel broke, or the alert is unactionable.""" + payload = _payload( + profiles={ + "web": _profile(), + "db": _profile(healthy=False, health_error="port 23335 is not listening"), + } + ) + code, body = health_response(payload) + assert code == 503 + assert body["status"] == "degraded" + assert body["unhealthy"] == ["db"] + assert body["errors"] == {"db": "port 23335 is not listening"} + + +def test_healthz_starting_is_not_reported_as_down() -> None: + """No health check yet must not fire a false alert on every restart. + + A fresh ``ponte start`` waits up to ``check_interval`` for its first + answer; calling that "down" would page someone on every restart. + """ + code, body = health_response(_payload(profiles={"web": _profile(healthy=None)})) + assert code == 200 + assert body["status"] == "starting" + + +def test_healthz_ok_when_no_profile_reported_at_all() -> None: + code, body = health_response(_payload(profiles={})) + assert code == 200 + assert body["status"] == "starting" + + +def test_healthz_survives_a_malformed_payload() -> None: + """The status file is written by whichever daemon version is installed.""" + code, body = health_response({"running": True, "profiles": "nonsense"}) + assert code == 200 + assert body["status"] == "starting" + + +# --------------------------------------------------------------------------- +# /metrics +# --------------------------------------------------------------------------- + + +def test_metrics_exposes_the_profile_state() -> None: + families = _parse_families(render_metrics(_payload(), now=_NOW)) + + assert _sample(families, "ponte_up") == "ponte_up 1" + assert _sample(families, "ponte_profiles_configured") == ( + "ponte_profiles_configured 1" + ) + assert _sample(families, "ponte_profiles_unhealthy") == ( + "ponte_profiles_unhealthy 0" + ) + assert _sample(families, "ponte_build_info") == ( + f'ponte_build_info{{version="{__version__}"}} 1' + ) + assert _sample(families, "ponte_profile_healthy", profile="web") == ( + 'ponte_profile_healthy{profile="web"} 1' + ) + assert _sample(families, "ponte_profile_sessions_total", profile="web") == ( + 'ponte_profile_sessions_total{profile="web"} 3' + ) + # Identity travels in one info metric, so a legend can show the server + # without repeating the string on every sample. + assert _sample( + families, + "ponte_profile_info", + profile="web", + destination="deploy@example.com:22", + ) == ( + 'ponte_profile_info{profile="web",destination="deploy@example.com:22"} 1' + ) + + +def test_metrics_session_age_is_computed_live() -> None: + families = _parse_families(render_metrics(_payload(), now=_NOW)) + assert _sample(families, "ponte_profile_session_uptime_seconds", profile="web") == ( + 'ponte_profile_session_uptime_seconds{profile="web"} 120.000' + ) + + +def test_metrics_reports_port_listening_state() -> None: + """The strongest signal of the lot: is the port *actually* forwarded?""" + families = _parse_families(render_metrics(_payload(), now=_NOW)) + assert _sample( + families, "ponte_profile_port_listening", profile="web", kind="remote", port="23334" + ) == ( + 'ponte_profile_port_listening{profile="web",kind="remote",port="23334"} 1' + ) + assert _sample( + families, "ponte_profile_port_listening", profile="web", kind="local", port="1080" + ) == 'ponte_profile_port_listening{profile="web",kind="local",port="1080"} 0' + + +def test_metrics_omits_unknown_values_instead_of_exporting_nan() -> None: + """A gap in a graph says "no data"; a NaN line invites a guess.""" + payload = _payload( + profiles={ + "web": _profile( + healthy=None, + process_alive=None, + current_session_at=None, + availability=None, + sessions_total=None, + last_disconnect_at=None, + last_notification_at=None, + remote_ports={}, + local_ports={}, + ) + } + ) + text = render_metrics(payload, now=_NOW) + for absent in ( + "ponte_profile_healthy", + "ponte_profile_process_alive", + "ponte_profile_session_uptime_seconds", + "ponte_profile_availability_ratio", + "ponte_profile_sessions_total", + "ponte_profile_port_listening", + ): + assert absent not in text + assert "ponte_up 1" in text + + +def test_metrics_when_daemon_is_stopped_still_answers_200_up_zero() -> None: + """The scrape must keep working while the thing it measures is down.""" + families = _parse_families(render_metrics({"running": False})) + assert _sample(families, "ponte_up") == "ponte_up 0" + assert "ponte_profile_sessions_total" not in families + + +def test_metrics_escapes_label_values() -> None: + """A quote or newline in a name must not forge or split a label. + + Profile names come from the config and destinations from ``[ssh]``; a real + newline leaking into a label would break the sample across two lines and + corrupt the whole scrape, not just that metric. + """ + name = 'we"b\n' + destination = 'host"x\\y\nz' + families = _parse_families(render_metrics(_payload(profiles={name: _profile(destination=destination)}), now=_NOW)) + + healthy = _sample(families, "ponte_profile_healthy") + assert healthy is not None + assert _unescape(_labels(healthy)["profile"]) == name + + info = _sample(families, "ponte_profile_info") + assert info is not None + assert _unescape(_labels(info)["destination"]) == destination + + +# --------------------------------------------------------------------------- +# The dashboard +# --------------------------------------------------------------------------- + + +def test_dashboard_shows_the_tunnel_state() -> None: + page = dashboard_html(_payload(), refresh=7, now=_NOW) + assert "web" in page + assert "deploy@example.com:22" in page + assert "2m 0s" in page # current session duration + assert "90.0%" in page # availability + assert "监听中" in page and "未监听" in page + assert "ssh exited with code 255" in page + assert '' in page + + +def test_dashboard_escapes_everything_from_outside() -> None: + """Profile names come from config and reasons from SSH's stderr.""" + payload = _payload( + profiles={ + "": _profile( + destination='">', + last_disconnect_reason="sshd said no", + recent_events=[ + {"at": _NOW, "type": "disconnected", "reason": "boom"} + ], + ) + } + ) + page = dashboard_html(payload, now=_NOW) + assert "" not in page + assert "<script>alert(1)</script>" in page + assert "sshd" not in page + assert "boom" not in page + + +def test_dashboard_reports_a_stopped_daemon() -> None: + page = dashboard_html({"running": False}, now=_NOW) + assert "守护进程未运行" in page + assert "ponte start" in page + + +def test_dashboard_waits_quietly_for_the_first_check() -> None: + page = dashboard_html(_payload(profiles={"web": _profile(healthy=None)}), now=_NOW) + assert "等待首次检查" in page + + +# --------------------------------------------------------------------------- +# End to end: a real server, real HTTP +# --------------------------------------------------------------------------- + + +@contextlib.contextmanager +def _running_server(provider, **kwargs) -> Iterator[str]: + """Start a server on an OS-assigned port and yield its base URL.""" + server = create_server(provider, host="127.0.0.1", port=0, **kwargs) + thread = threading.Thread( + target=server.serve_forever, kwargs={"poll_interval": 0.05}, daemon=True + ) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}" + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + + +def _get(url: str, *, headers: dict[str, str] | None = None, method: str = "GET"): + """Return ``(status, headers, body_text)`` for *url* without raising.""" + request = urllib.request.Request(url, headers=headers or {}, method=method) + try: + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, response.headers, response.read().decode("utf-8") + except urllib.error.HTTPError as error: + return error.code, error.headers, error.read().decode("utf-8") + + +def test_end_to_end_endpoints_answer() -> None: + with _running_server(lambda: _payload()) as base: + status, headers, body = _get(base + "/") + assert status == 200 + assert headers["Content-Type"].startswith("text/html") + assert "web" in body + + status, _, body = _get(base + "/healthz") + assert status == 200 + assert json.loads(body)["status"] == "ok" + + status, headers, body = _get(base + "/metrics") + assert status == 200 + assert headers["Content-Type"].startswith("text/plain") + assert "ponte_up 1" in body + + status, _, body = _get(base + "/status.json") + assert status == 200 + assert json.loads(body)["profiles"]["web"]["sessions_total"] == 3 + + +def test_end_to_end_healthz_tracks_the_tunnel() -> None: + """The probe reports the tunnel, not the process: 503 when it breaks.""" + state = {"payload": _payload()} + with _running_server(lambda: state["payload"]) as base: + assert _get(base + "/healthz")[0] == 200 + state["payload"] = _payload( + profiles={"web": _profile(healthy=False, health_error="port closed")} + ) + status, _, body = _get(base + "/healthz") + assert status == 503 + assert json.loads(body)["errors"] == {"web": "port closed"} + + +def test_end_to_end_status_is_never_cached() -> None: + """A replayed "healthy" page is the exact failure the endpoint prevents.""" + state = {"payload": _payload()} + with _running_server(lambda: state["payload"]) as base: + _, headers, _ = _get(base + "/") + assert headers["Cache-Control"] == "no-store" + state["payload"] = {"running": False} + assert "ponte start" in _get(base + "/")[2] + + +def test_end_to_end_unknown_path_and_method() -> None: + with _running_server(lambda: _payload()) as base: + status, _, body = _get(base + "/nope") + assert status == 404 + assert "/metrics" in json.loads(body)["endpoints"] + + status, headers, body = _get(base + "/", method="POST") + assert status == 405 + assert headers["Allow"] == "GET, HEAD" + assert "read-only" in json.loads(body)["error"] + + +def test_end_to_end_head_sends_headers_without_a_body() -> None: + with _running_server(lambda: _payload()) as base: + status, headers, body = _get(base + "/metrics", method="HEAD") + assert status == 200 + assert int(headers["Content-Length"]) > 0 + assert body == "" + + +def test_end_to_end_token_gate() -> None: + """With a token set, nothing is served without it — header or query.""" + with _running_server(lambda: _payload(), token="s3cret") as base: + for path in _ROUTES_FOR_TOKEN_TEST: + status, headers, _ = _get(base + path) + assert status == 401, path + assert headers["WWW-Authenticate"].startswith("Bearer") + + assert _get(base + "/healthz?token=s3cret")[0] == 200 + assert _get(base + "/", headers={"Authorization": "Bearer s3cret"})[0] == 200 + assert _get(base + "/healthz?token=wrong")[0] == 401 + + +_ROUTES_FOR_TOKEN_TEST = ("/", "/healthz", "/metrics", "/status.json") + + +def test_end_to_end_provider_failure_answers_500_without_dropping_the_client() -> None: + def broken() -> dict: + raise RuntimeError("status file is a mess") + + with _running_server(broken) as base: + status, _, body = _get(base + "/healthz") + assert status == 500 + assert "status unavailable" in json.loads(body)["error"] + + +# --------------------------------------------------------------------------- +# Bind safety and URL rendering +# --------------------------------------------------------------------------- + + +def test_serve_refuses_to_expose_without_a_token() -> None: + """Refusing beats warning: the page maps your infrastructure.""" + with pytest.raises(ConfigValidationError) as caught: + create_server(lambda: _payload(), host="0.0.0.0", port=0) + assert "token" in str(caught.value) + + +def test_serve_allows_a_non_loopback_bind_when_a_token_is_set() -> None: + """The gate is the token, not the address: with one, exposing is allowed. + + Checked through :func:`ensure_bindable` rather than by actually binding a + public interface — a test has no business opening a port the whole LAN can + reach (and on Windows that alone can raise a firewall prompt). + """ + ensure_bindable("0.0.0.0", "s3cret") # does not raise + with pytest.raises(ConfigValidationError): + ensure_bindable("0.0.0.0", "") + + +def test_create_server_carries_its_settings() -> None: + server = create_server( + lambda: _payload(), host="127.0.0.1", port=0, token="s3cret", refresh=9 + ) + try: + assert server.token == "s3cret" + assert server.refresh == 9 + finally: + server.server_close() + + +@pytest.mark.parametrize( + ("host", "expected"), + [ + ("", "http://127.0.0.1:8787/"), + ("127.0.0.1", "http://127.0.0.1:8787/"), + ("0.0.0.0", "http://0.0.0.0:8787/"), + ("::1", "http://[::1]:8787/"), + ("192.168.1.5", "http://192.168.1.5:8787/metrics"), + ], +) +def test_serve_url_formats_the_link(host: str, expected: str) -> None: + path = "/metrics" if host == "192.168.1.5" else "/" + assert serve_url(host, 8787, path) == expected