give --watch a per-tick verdict and optional webhook alerts

watch.py never imported interpret and had no notion of a verdict at all -
summarize_cycle() now runs the same latency_findings/speed_findings/
overall_verdict pipeline the one-shot report uses, so every tick carries
status/score/finding_ids and the dashboard caption shows a verdict line.
Real improvement on its own, independent of whether a webhook exists.

New alerting.py carries the whole notification contract as pure functions:
should_fire() only fires on an actual transition, into a state the user
opted into via watch.webhook_on, debounced by
watch.webhook_min_interval_seconds (floored at 30 in a validator) so a
flapping link can't spam an endpoint every tick. Recovery is its own
opt-in "recovered" token, separate from configuring "crit", since wanting
outage alerts doesn't imply wanting a second ping when they clear.
build_payload() is flat JSON, not shaped for Slack/Discord - that's the
receiver's problem. post_webhook() swallows every failure and returns a
bool; a dead endpoint must never interrupt the loop watching it.

New --webhook URL flag overrides watch.webhook_url per run.

Verified the tick sequence end-to-end outside the Rich Live loop -
sending real OS signals to test --watch's Ctrl-C handling in this shell
wasn't reliable, so exercised should_fire/build_payload/render_dashboard/
to_report in the same order run_watch calls them.

Part of plans/zesty-zooming-wilkinson.md (gitignored).
This commit is contained in:
Andrew Pimenov 2026-08-12 22:26:31 +03:00
parent 245533f904
commit e27ccdfab5
12 changed files with 284 additions and 7 deletions

View file

@ -23,6 +23,7 @@ src/netsleuth/
speed.py Cascading speedtest, throughput math, cfL4 header parsing, bufferbloat probe. speed.py Cascading speedtest, throughput math, cfL4 header parsing, bufferbloat probe.
exporter.py JSON and Markdown rendering, atomic writes into ./logs/. Which exporter.py JSON and Markdown rendering, atomic writes into ./logs/. Which
artifacts get written is decided by the caller (cli.py), not here. artifacts get written is decided by the caller (cli.py), not here.
alerting.py Pure webhook-transition contract (should_fire debounce) plus a plain httpx POST for --watch alerts.
compare.py --compare: diff two saved JSON reports; render_diff_brief() feeds the automatic vs-previous-run summary. compare.py --compare: diff two saved JSON reports; render_diff_brief() feeds the automatic vs-previous-run summary.
history.py Same-ASN report lookup by filename convention: report_key(), find_previous(), latest_key(). history.py Same-ASN report lookup by filename convention: report_key(), find_previous(), latest_key().
trend.py --trend N: aggregates N saved JSON reports into per-label sparkline series. trend.py --trend N: aggregates N saved JSON reports into per-label sparkline series.
@ -154,6 +155,12 @@ The AS prefix benchmark (`probes/prefix_benchmark.py`) gets the same treatment f
**`--trend` needs no live network call to find "this network's" history.** Rather than re-running identity detection just to know which ASN's reports to load, `history.latest_key()` reads it back off the newest `report_*.json` filename already on disk — the same file-naming convention already encodes it, and a read-only historical view has no reason to touch the network at all. **`--trend` needs no live network call to find "this network's" history.** Rather than re-running identity detection just to know which ASN's reports to load, `history.latest_key()` reads it back off the newest `report_*.json` filename already on disk — the same file-naming convention already encodes it, and a read-only historical view has no reason to touch the network at all.
**`--watch` had no verdict at all before this, and giving it one came first.** `watch.py::run_watch()` only ever fanned out pings (plus an occasional speedtest) and rendered raw numbers — `interpret` was never imported. Wiring alerting to "a verdict transition" was impossible until there was a verdict to transition between, so `summarize_cycle()` now reuses the exact same pure `latency_findings()`/`speed_findings()`/`overall_verdict()` functions the one-shot report already uses, gaining `status`/`score`/`finding_ids` on every tick — a real improvement to the dashboard on its own, independent of whether a webhook is ever configured.
**The whole webhook debounce contract lives in one pure function.** `alerting.should_fire(previous, current, last_fired_at, now, min_interval_s, fire_on)` decides everything: fires only on an actual transition (never on the first tick, since there's nothing to transition from), only into a state the user opted into (`watch.webhook_on`), and only after `webhook_min_interval_seconds` (floored at 30 in a validator) has elapsed since the last firing — a flapping link cannot spam the endpoint every tick. Recovery (`crit`/`warn` → `ok`/`info`) is its own opt-in token (`"recovered"`), not implied by configuring `"crit"`, because a user who wants to know about outages doesn't necessarily want a second ping when they end.
**The webhook payload is a flat JSON body, not shaped for any particular chat platform.** `alerting.build_payload()` returns `{tool, asn, at, previous, current, score, findings, host}` — Slack/Discord-specific formatting is a concern for whatever receives the POST, not for netsleuth. `post_webhook()` swallows every `httpx`/`OSError` failure and returns `False`; a dead webhook endpoint must never interrupt the monitoring loop it's supposed to be watching.
**Python 3.14 is the floor, not 3.10.** The project moved off `>=3.10` when this batch of probes was added; none of the new code actually depends on a 3.11+-only construct (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — the two-sequential-connection TLS design above was chosen independently of what the interpreter allows. The version floor moved because there was no longer a reason to hold it back, not because a specific new feature required it. **Python 3.14 is the floor, not 3.10.** The project moved off `>=3.10` when this batch of probes was added; none of the new code actually depends on a 3.11+-only construct (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — the two-sequential-connection TLS design above was chosen independently of what the interpreter allows. The version floor moved because there was no longer a reason to hold it back, not because a specific new feature required it.
## Storage ## Storage
@ -172,4 +179,4 @@ Five sections back the L4-L7 probes, all opt-in at the CLI level and all with a
## Tests ## Tests
`uv run pytest -q`. Roughly 807 tests covering the parsers, normalizers, verdict engine, DNSBL decoding, throughput math, ping statistics, jitter percentiles, serialization and the report diff. Glue is deliberately untested: live HTTP, real ICMP, Typer wiring, `psutil`/`ctypes` calls and the `--watch` timing loop. Traceroute fixtures live in `tests/fixtures/traceroute/{windows,linux,darwin}/` and include a cp866-encoded Russian Windows sample, because that encoding is exactly what breaks naive parsers. DoH response fixtures (RFC 8484 JSON form) live in `tests/fixtures/api/doh_*.json` alongside the other provider fixtures. `uv run pytest -q`. Roughly 825 tests covering the parsers, normalizers, verdict engine, DNSBL decoding, throughput math, ping statistics, jitter percentiles, serialization and the report diff. Glue is deliberately untested: live HTTP, real ICMP, Typer wiring, `psutil`/`ctypes` calls and the `--watch` timing loop. Traceroute fixtures live in `tests/fixtures/traceroute/{windows,linux,darwin}/` and include a cp866-encoded Russian Windows sample, because that encoding is exactly what breaks naive parsers. DoH response fixtures (RFC 8484 JSON form) live in `tests/fixtures/api/doh_*.json` alongside the other provider fixtures.

View file

@ -23,6 +23,7 @@ src/netsleuth/
speed.py Каскадный замер скорости, математика пропускной способности, разбор cfL4, bufferbloat. speed.py Каскадный замер скорости, математика пропускной способности, разбор cfL4, bufferbloat.
exporter.py Рендер JSON и Markdown, атомарная запись в ./logs/. Какие exporter.py Рендер JSON и Markdown, атомарная запись в ./logs/. Какие
артефакты писать, решает вызывающий код (cli.py), не этот модуль. артефакты писать, решает вызывающий код (cli.py), не этот модуль.
alerting.py Чистый контракт перехода вебхука (дебаунс should_fire) плюс обычный httpx POST для алертов --watch.
compare.py --compare: сравнение двух сохранённых JSON-отчётов; render_diff_brief() питает автоматическую сводку vs-прошлый-прогон. compare.py --compare: сравнение двух сохранённых JSON-отчётов; render_diff_brief() питает автоматическую сводку vs-прошлый-прогон.
history.py Поиск отчётов той же ASN по соглашению об именах файлов: report_key(), find_previous(), latest_key(). history.py Поиск отчётов той же ASN по соглашению об именах файлов: report_key(), find_previous(), latest_key().
trend.py --trend N: агрегирует N сохранённых JSON-отчётов в спарклайн-серии по меткам. trend.py --trend N: агрегирует N сохранённых JSON-отчётов в спарклайн-серии по меткам.
@ -154,6 +155,12 @@ src/netsleuth/
**`--trend` не требует живого сетевого запроса, чтобы понять «историю этой сети».** Вместо повторного запуска определения identity только чтобы узнать, чьи отчёты грузить, `history.latest_key()` считывает это обратно из имени самого нового `report_*.json` на диске — то же соглашение об именовании файлов уже это кодирует, а read-only исторический просмотр не имеет причин вообще трогать сеть. **`--trend` не требует живого сетевого запроса, чтобы понять «историю этой сети».** Вместо повторного запуска определения identity только чтобы узнать, чьи отчёты грузить, `history.latest_key()` считывает это обратно из имени самого нового `report_*.json` на диске — то же соглашение об именовании файлов уже это кодирует, а read-only исторический просмотр не имеет причин вообще трогать сеть.
**У `--watch` вообще не было вердикта до этого, и дать ему вердикт было первым шагом.** `watch.py::run_watch()` только веером гонял пинги (плюс изредка спидтест) и рендерил сырые числа — `interpret` никогда не импортировался. Подключить алертинг к «переходу вердикта» было невозможно, пока не было вердикта, между которым переходить, так что `summarize_cycle()` теперь переиспользует те же самые чистые функции `latency_findings()`/`speed_findings()`/`overall_verdict()`, что уже использует одноразовый отчёт, получая `status`/`score`/`finding_ids` на каждый тик — реальное улучшение дашборда само по себе, независимо от того, настроен ли когда-либо вебхук.
**Весь контракт дебаунса вебхука живёт в одной чистой функции.** `alerting.should_fire(previous, current, last_fired_at, now, min_interval_s, fire_on)` решает всё: срабатывает только на реальном переходе (никогда на первом тике, поскольку не с чем сравнивать), только в состояние, на которое пользователь подписался (`watch.webhook_on`), и только после того как прошло `webhook_min_interval_seconds` (пол в 30 в валидаторе) с последнего срабатывания — флапающий канал не может спамить эндпоинт на каждом тике. Восстановление (`crit`/`warn` → `ok`/`info`) — отдельный опциональный токен (`"recovered"`), не подразумеваемый настройкой `"crit"`, потому что пользователь, который хочет знать об отключениях, не обязательно хочет второй пинг, когда они кончаются.
**Полезная нагрузка вебхука — плоское JSON-тело, не заточенное ни под какую конкретную чат-платформу.** `alerting.build_payload()` возвращает `{tool, asn, at, previous, current, score, findings, host}` — форматирование под Slack/Discord — забота того, что принимает POST, а не netsleuth. `post_webhook()` глотает любой отказ `httpx`/`OSError` и возвращает `False`; мёртвый эндпоинт вебхука не должен прерывать цикл мониторинга, за которым он должен следить.
**Python 3.14 — минимальная версия, не 3.10.** Проект ушёл с `>=3.10` одновременно с этим набором проб; ни один новый код на деле не зависит от конструкций только для 3.11+ (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — схема TLS с двумя последовательными соединениями выше выбрана независимо от того, что позволяет интерпретатор. Планка версии сдвинулась потому, что не осталось причин её удерживать, а не потому, что это потребовалось для конкретной новой функции. **Python 3.14 — минимальная версия, не 3.10.** Проект ушёл с `>=3.10` одновременно с этим набором проб; ни один новый код на деле не зависит от конструкций только для 3.11+ (`ssl.SSLObject.start_tls()`, `asyncio.timeout()`, `datetime.UTC`) — схема TLS с двумя последовательными соединениями выше выбрана независимо от того, что позволяет интерпретатор. Планка версии сдвинулась потому, что не осталось причин её удерживать, а не потому, что это потребовалось для конкретной новой функции.
## Хранение ## Хранение
@ -172,4 +179,4 @@ src/netsleuth/
## Тесты ## Тесты
`uv run pytest -q`. Около 807 тестов: парсеры, нормализаторы, движок вердиктов, декодирование DNSBL, математика пропускной способности, статистика пингов, перцентили джиттера, сериализация и сравнение отчётов. Клей намеренно не тестируется: живой HTTP, настоящий ICMP, обвязка Typer, вызовы `psutil`/`ctypes` и цикл `--watch`. Фикстуры трассировок лежат в `tests/fixtures/traceroute/{windows,linux,darwin}/` и включают образец русской Windows в cp866 — именно эта кодировка ломает наивные парсеры. Фикстуры DoH-ответов (формат RFC 8484 JSON) лежат в `tests/fixtures/api/doh_*.json` рядом с фикстурами других провайдеров. `uv run pytest -q`. Около 825 тестов: парсеры, нормализаторы, движок вердиктов, декодирование DNSBL, математика пропускной способности, статистика пингов, перцентили джиттера, сериализация и сравнение отчётов. Клей намеренно не тестируется: живой HTTP, настоящий ICMP, обвязка Typer, вызовы `psutil`/`ctypes` и цикл `--watch`. Фикстуры трассировок лежат в `tests/fixtures/traceroute/{windows,linux,darwin}/` и включают образец русской Windows в cp866 — именно эта кодировка ломает наивные парсеры. Фикстуры DoH-ответов (формат RFC 8484 JSON) лежат в `tests/fixtures/api/doh_*.json` рядом с фикстурами других провайдеров.

View file

@ -80,7 +80,8 @@ None of this is required — netsleuth degrades gracefully without any of it —
| `--target AS64500` / `--target 64500` / `--target 1.2.3.4` / `--target example.com` | Investigate a specific network instead of your own connection — the `AS` prefix on a bare ASN is optional | | `--target AS64500` / `--target 64500` / `--target 1.2.3.4` / `--target example.com` | Investigate a specific network instead of your own connection — the `AS` prefix on a bare ASN is optional |
| `--target-host <ip\|domain>` | Add an extra host to the ping and traceroute fan-out | | `--target-host <ip\|domain>` | Add an extra host to the ping and traceroute fan-out |
| `--speedtest-server <id\|url>` | Pin the speedtest to a specific server | | `--speedtest-server <id\|url>` | Pin the speedtest to a specific server |
| `--watch` | Continuous monitoring with a live dashboard — for catching intermittent drops. Writes one `logs/watch_*.json` time series per session, not one report per tick | | `--watch` | Continuous monitoring with a live dashboard — for catching intermittent drops. Writes one `logs/watch_*.json` time series per session, not one report per tick. Each tick gets a verdict (same scoring engine as a one-shot report) shown in the dashboard caption |
| `--webhook <url>` | With `--watch`, POST a JSON alert to this URL on a verdict transition (into `warn`/`crit`, or into recovery if `watch.webhook_on` includes `"recovered"`). Debounced by `watch.webhook_min_interval_seconds` (floor 30s) so a flapping link can't spam it. Overrides `watch.webhook_url` in `config.yaml` |
| `--compare a.json b.json` | Diff two earlier reports: before/after a VPN switch, before/after an ISP call. Read-only — no probing, no network calls. Reads reports produced with `--format json` | | `--compare a.json b.json` | Diff two earlier reports: before/after a VPN switch, before/after an ISP call. Read-only — no probing, no network calls. Reads reports produced with `--format json` |
| `--trend N` | Sparkline trend across the last N saved JSON reports for this network. Read-only — resolves which network from the newest report already on disk, no probing needed | | `--trend N` | Sparkline trend across the last N saved JSON reports for this network. Read-only — resolves which network from the newest report already on disk, no probing needed |
| `--no-auto-diff` | Skip the one-line vs-previous-run summary a run otherwise prints automatically when it writes JSON and an earlier JSON report for the same network exists | | `--no-auto-diff` | Skip the one-line vs-previous-run summary a run otherwise prints automatically when it writes JSON and an earlier JSON report for the same network exists |

View file

@ -80,7 +80,8 @@ netsleuth
| `--target AS64500` / `--target 64500` / `--target 1.2.3.4` / `--target example.com` | Исследовать конкретную сеть вместо своего подключения — префикс `AS` перед голым номером ASN необязателен | | `--target AS64500` / `--target 64500` / `--target 1.2.3.4` / `--target example.com` | Исследовать конкретную сеть вместо своего подключения — префикс `AS` перед голым номером ASN необязателен |
| `--target-host <ip\|domain>` | Добавить хост в веер пингов и трассировок | | `--target-host <ip\|domain>` | Добавить хост в веер пингов и трассировок |
| `--speedtest-server <id\|url>` | Зафиксировать конкретный сервер замера скорости | | `--speedtest-server <id\|url>` | Зафиксировать конкретный сервер замера скорости |
| `--watch` | Непрерывный мониторинг с живой панелью — чтобы поймать плавающие обрывы. Пишет один тайм-ряд `logs/watch_*.json` за сессию, а не отдельный отчёт на каждый тик | | `--watch` | Непрерывный мониторинг с живой панелью — чтобы поймать плавающие обрывы. Пишет один тайм-ряд `logs/watch_*.json` за сессию, а не отдельный отчёт на каждый тик. Каждый тик получает вердикт (тот же движок оценки, что и в одноразовом отчёте), показанный в подписи панели |
| `--webhook <url>` | Вместе с `--watch` отправляет POST с JSON-алертом на этот URL при переходе вердикта (в `warn`/`crit`, либо при восстановлении, если `watch.webhook_on` включает `"recovered"`). С дебаунсом через `watch.webhook_min_interval_seconds` (пол 30с), чтобы флапающий канал не мог его заспамить. Переопределяет `watch.webhook_url` в `config.yaml` |
| `--compare a.json b.json` | Сравнить два прошлых отчёта: до/после смены VPN, до/после звонка провайдеру. Только чтение — без проб и сетевых запросов. Читает отчёты, полученные с `--format json` | | `--compare a.json b.json` | Сравнить два прошлых отчёта: до/после смены VPN, до/после звонка провайдеру. Только чтение — без проб и сетевых запросов. Читает отчёты, полученные с `--format json` |
| `--trend N` | Спарклайн-тренд по последним N сохранённым JSON-отчётам для этой сети. Только чтение — сеть определяется по самому новому отчёту уже на диске, проб не требуется | | `--trend N` | Спарклайн-тренд по последним N сохранённым JSON-отчётам для этой сети. Только чтение — сеть определяется по самому новому отчёту уже на диске, проб не требуется |
| `--no-auto-diff` | Пропустить однострочную сводку vs-прошлый-прогон, которую прогон иначе печатает автоматически, когда пишет JSON и есть более ранний JSON-отчёт для той же сети | | `--no-auto-diff` | Пропустить однострочную сводку vs-прошлый-прогон, которую прогон иначе печатает автоматически, когда пишет JSON и есть более ранний JSON-отчёт для той же сети |

View file

@ -152,3 +152,9 @@ watch:
interval_seconds: 60 interval_seconds: 60
speedtest_every_n_cycles: 10 speedtest_every_n_cycles: 10
dashboard_refresh_hz: 4 dashboard_refresh_hz: 4
# Optional: POST a JSON alert on a verdict transition during --watch. Overridable
# per run with --webhook. min_interval_seconds is floored at 30 in code so a
# flapping link cannot fire the webhook every tick.
webhook_url: null
webhook_min_interval_seconds: 300
webhook_on: ["warn", "crit"]

51
src/netsleuth/alerting.py Normal file
View file

@ -0,0 +1,51 @@
from __future__ import annotations
import httpx
_BAD_STATUSES = {"warn", "crit"}
_GOOD_STATUSES = {"ok", "info"}
def should_fire(
previous: str | None,
current: str,
last_fired_at: float | None,
now: float,
min_interval_s: float,
fire_on: set[str],
) -> bool:
if previous is None or previous == current:
return False
if current in _BAD_STATUSES:
event = current
elif previous in _BAD_STATUSES and current in _GOOD_STATUSES:
event = "recovered"
else:
return False
if event not in fire_on:
return False
if last_fired_at is not None and (now - last_fired_at) < min_interval_s:
return False
return True
def build_payload(session_meta: dict, cycle_summary: dict, previous: str | None, current: str) -> dict:
return {
"tool": "netsleuth",
"asn": session_meta.get("asn"),
"at": cycle_summary.get("at"),
"previous": previous,
"current": current,
"score": cycle_summary.get("score"),
"findings": cycle_summary.get("finding_ids") or [],
"host": session_meta.get("interface"),
}
async def post_webhook(client: httpx.AsyncClient, url: str, payload: dict, timeout: float) -> bool:
try:
response = await client.post(url, json=payload, timeout=timeout)
response.raise_for_status()
return True
except (httpx.HTTPError, OSError):
return False

View file

@ -93,6 +93,7 @@ class Options:
formats: frozenset[str] = frozenset({"md"}) formats: frozenset[str] = frozenset({"md"})
interface: str | None = None interface: str | None = None
bind: BindTarget | None = None bind: BindTarget | None = None
webhook: str | None = None
def parse_formats( def parse_formats(
@ -764,6 +765,11 @@ def run(
target_host: Optional[str] = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."), target_host: Optional[str] = typer.Option(None, "--target-host", help="Extra host for ping/traceroute."),
speedtest_server: Optional[str] = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."), speedtest_server: Optional[str] = typer.Option(None, "--speedtest-server", help="Pin the speedtest server."),
watch: bool = typer.Option(False, "--watch", help="Continuous monitoring with a live dashboard."), watch: bool = typer.Option(False, "--watch", help="Continuous monitoring with a live dashboard."),
webhook: Optional[str] = typer.Option(
None,
"--webhook",
help="With --watch, POST a JSON alert to this URL on a verdict transition (overrides watch.webhook_url).",
),
compare: Optional[Tuple[Path, Path]] = typer.Option( compare: Optional[Tuple[Path, Path]] = typer.Option(
None, "--compare", help="Diff two saved JSON reports — reads reports produced with --format json." None, "--compare", help="Diff two saved JSON reports — reads reports produced with --format json."
), ),
@ -863,6 +869,7 @@ def run(
prefix_bench=prefix_bench, prefix_bench=prefix_bench,
dpi_target=my_server, dpi_target=my_server,
formats=formats, formats=formats,
webhook=webhook,
) )
if interface: if interface:
addrs_by_iface = { addrs_by_iface = {

View file

@ -293,6 +293,35 @@ class Watch(BaseModel):
interval_seconds: int = 60 interval_seconds: int = 60
speedtest_every_n_cycles: int = 10 speedtest_every_n_cycles: int = 10
dashboard_refresh_hz: int = 4 dashboard_refresh_hz: int = 4
webhook_url: str | None = None
webhook_min_interval_seconds: int = 300
webhook_on: list[str] = Field(default_factory=lambda: ["warn", "crit"])
@field_validator("webhook_min_interval_seconds")
@classmethod
def _min_interval_floor(cls, value: int) -> int:
if value < 30:
raise ValueError(
"watch.webhook_min_interval_seconds may not be below 30 "
"(a flapping link must not be able to fire the webhook every tick)"
)
return value
@field_validator("webhook_on")
@classmethod
def _validate_webhook_on(cls, value: list[str]) -> list[str]:
allowed = {"ok", "info", "warn", "crit", "recovered"}
unknown = sorted(set(value) - allowed)
if unknown:
raise ValueError(f"watch.webhook_on has unknown entries {unknown}; expected a subset of {sorted(allowed)}")
return value
@field_validator("webhook_url")
@classmethod
def _validate_webhook_url(cls, value: str | None) -> str | None:
if value is not None and not value.startswith(("http://", "https://")):
raise ValueError("watch.webhook_url must start with http:// or https://")
return value
class Settings(BaseSettings): class Settings(BaseSettings):

View file

@ -12,8 +12,10 @@ from rich.live import Live
from rich.table import Table from rich.table import Table
from netsleuth import __version__ from netsleuth import __version__
from netsleuth.config import Settings from netsleuth.alerting import build_payload, post_webhook, should_fire
from netsleuth.config import Settings, Thresholds
from netsleuth.exporter import atomic_write, compact_timestamp, dump_json, sanitize_name, sparkline from netsleuth.exporter import atomic_write, compact_timestamp, dump_json, sanitize_name, sparkline
from netsleuth.interpret import latency_findings, overall_verdict, speed_findings
from netsleuth.ip_geo import gather_identity from netsleuth.ip_geo import gather_identity
from netsleuth.models import PingResult, SpeedResult, to_jsonable from netsleuth.models import PingResult, SpeedResult, to_jsonable
from netsleuth.netinfo import detect_capabilities from netsleuth.netinfo import detect_capabilities
@ -41,7 +43,13 @@ def summarize_cycle(
at: str, at: str,
pings: list[PingResult], pings: list[PingResult],
speed: SpeedResult | None, speed: SpeedResult | None,
thresholds: Thresholds | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
thresholds = thresholds or Thresholds()
findings = latency_findings(pings, thresholds) + (
speed_findings(speed, thresholds.bufferbloat_ms) if speed is not None else []
)
status, score, _summary = overall_verdict(findings)
return { return {
"cycle": cycle, "cycle": cycle,
"at": at, "at": at,
@ -63,6 +71,9 @@ def summarize_cycle(
"upload_mbps": speed.upload_mbps, "upload_mbps": speed.upload_mbps,
"bufferbloat_grade": speed.bufferbloat_grade, "bufferbloat_grade": speed.bufferbloat_grade,
}, },
"status": status,
"score": score,
"finding_ids": [f.id for f in findings],
} }
@ -75,6 +86,7 @@ class WatchSession:
interface: str | None = None interface: str | None = None
bind_ipv4: str | None = None bind_ipv4: str | None = None
cycles: list[dict] = field(default_factory=list) cycles: list[dict] = field(default_factory=list)
alerts_fired: int = 0
def add(self, summary: dict) -> None: def add(self, summary: dict) -> None:
self.cycles.append(summary) self.cycles.append(summary)
@ -106,6 +118,7 @@ class WatchSession:
"cycle_count": len(self.cycles), "cycle_count": len(self.cycles),
"interface": self.interface, "interface": self.interface,
"bind_ipv4": self.bind_ipv4, "bind_ipv4": self.bind_ipv4,
"alerts_fired": self.alerts_fired,
}, },
"cycles": to_jsonable(self.cycles), "cycles": to_jsonable(self.cycles),
} }
@ -129,12 +142,18 @@ def render_dashboard(session: WatchSession) -> Table:
f"{row.get('loss_pct', 0)}%", f"{row.get('loss_pct', 0)}%",
sparkline(session.history(label)[-40:]), sparkline(session.history(label)[-40:]),
) )
caption_parts: list[str] = []
if session.cycles:
last = session.cycles[-1]
caption_parts.append(f"verdict: {last.get('status', '?')} ({last.get('score', '?')}/100)")
speed = next((c["speed"] for c in reversed(session.cycles) if c["speed"]), None) speed = next((c["speed"] for c in reversed(session.cycles) if c["speed"]), None)
if speed: if speed:
table.caption = ( caption_parts.append(
f"last speedtest: {speed['download_mbps']} / {speed['upload_mbps']} Mbps " f"last speedtest: {speed['download_mbps']} / {speed['upload_mbps']} Mbps "
f"via {speed['method']} · bufferbloat {speed['bufferbloat_grade']}" f"via {speed['method']} · bufferbloat {speed['bufferbloat_grade']}"
) )
if caption_parts:
table.caption = " · ".join(caption_parts)
return table return table
@ -158,6 +177,16 @@ async def run_watch(settings: Settings, options) -> Path:
console = Console() console = Console()
console.print(f"netsleuth {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop") console.print(f"netsleuth {__version__} · watching every {session.interval_seconds}s · Ctrl-C to stop")
webhook_url = options.webhook or settings.watch.webhook_url
fire_on = set(settings.watch.webhook_on)
if webhook_url:
console.print(
f"[yellow]--watch will POST your ASN and finding titles to {webhook_url} "
f"on a verdict transition into {sorted(fire_on)}.[/yellow]"
)
previous_status: str | None = None
last_fired_at: float | None = None
transport = httpx.AsyncHTTPTransport(local_address=bind.ipv4) if bind else None transport = httpx.AsyncHTTPTransport(local_address=bind.ipv4) if bind else None
async with httpx.AsyncClient( async with httpx.AsyncClient(
transport=transport, transport=transport,
@ -192,7 +221,27 @@ async def run_watch(settings: Settings, options) -> Path:
timeout=settings.timeouts.speedtest_seconds, timeout=settings.timeouts.speedtest_seconds,
) )
speed = result.data if result.status == "ok" else None speed = result.data if result.status == "ok" else None
session.add(summarize_cycle(cycle, utc_now_iso(), pings, speed)) cycle_summary = summarize_cycle(cycle, utc_now_iso(), pings, speed, settings.thresholds)
session.add(cycle_summary)
current_status = cycle_summary["status"]
if webhook_url and should_fire(
previous_status,
current_status,
last_fired_at,
time.perf_counter(),
settings.watch.webhook_min_interval_seconds,
fire_on,
):
payload = build_payload(
{"asn": session.asn, "interface": session.interface},
cycle_summary,
previous_status,
current_status,
)
if await post_webhook(client, webhook_url, payload, settings.timeouts.http_seconds):
last_fired_at = time.perf_counter()
session.alerts_fired += 1
previous_status = current_status
live.update(render_dashboard(session)) live.update(render_dashboard(session))
await asyncio.sleep(next_delay(began, time.perf_counter(), session.interval_seconds)) await asyncio.sleep(next_delay(began, time.perf_counter(), session.interval_seconds))
cycle += 1 cycle += 1

70
tests/test_alerting.py Normal file
View file

@ -0,0 +1,70 @@
from __future__ import annotations
import httpx
import pytest
from netsleuth.alerting import build_payload, post_webhook, should_fire
from netsleuth.exporter import dump_json
def test_should_fire_is_silent_on_the_first_ever_tick():
assert should_fire(None, "crit", None, 100.0, 300.0, {"crit"}) is False
def test_should_fire_is_silent_when_there_is_no_transition():
assert should_fire("crit", "crit", None, 100.0, 300.0, {"crit"}) is False
def test_should_fire_is_silent_when_the_new_state_is_not_in_fire_on():
assert should_fire("ok", "warn", None, 100.0, 300.0, {"crit"}) is False
def test_should_fire_respects_the_minimum_interval():
assert should_fire("ok", "crit", 100.0, 150.0, 300.0, {"crit"}) is False
def test_should_fire_fires_after_the_interval_elapses():
assert should_fire("ok", "crit", 100.0, 450.0, 300.0, {"crit"}) is True
def test_should_fire_fires_on_first_transition_with_no_prior_firing():
assert should_fire("ok", "crit", None, 100.0, 300.0, {"crit"}) is True
def test_should_fire_treats_recovery_as_a_separate_event():
assert should_fire("crit", "ok", None, 100.0, 300.0, {"crit"}) is False
assert should_fire("crit", "ok", None, 100.0, 300.0, {"crit", "recovered"}) is True
def test_should_fire_does_not_treat_info_to_ok_as_a_recovery():
assert should_fire("info", "ok", None, 100.0, 300.0, {"recovered"}) is False
def test_build_payload_shape_and_json_serializable():
payload = build_payload(
{"asn": "AS64500", "interface": "Ethernet"},
{"at": "2026-08-12T00:00:00Z", "score": 42, "finding_ids": ["latency.avg.a"]},
"ok",
"crit",
)
assert payload["tool"] == "netsleuth"
assert payload["asn"] == "AS64500"
assert payload["previous"] == "ok"
assert payload["current"] == "crit"
assert payload["score"] == 42
assert payload["findings"] == ["latency.avg.a"]
dump_json(payload) # must not raise
async def test_post_webhook_returns_true_on_success(httpx_mock):
httpx_mock.add_response(url="https://example.com/hook", status_code=200)
async with httpx.AsyncClient() as client:
ok = await post_webhook(client, "https://example.com/hook", {"a": 1}, timeout=5.0)
assert ok is True
async def test_post_webhook_returns_false_and_never_raises_on_failure(httpx_mock):
httpx_mock.add_response(url="https://example.com/hook", status_code=500)
async with httpx.AsyncClient() as client:
ok = await post_webhook(client, "https://example.com/hook", {"a": 1}, timeout=5.0)
assert ok is False

View file

@ -224,6 +224,34 @@ def test_tls_pinned_fingerprints_are_capped_at_32_entries(tmp_path):
load_settings(config_path=path, env_file=tmp_path / "nope.env") load_settings(config_path=path, env_file=tmp_path / "nope.env")
def test_watch_webhook_min_interval_rejects_below_the_floor(tmp_path):
path = tmp_path / "config.yaml"
path.write_text("watch:\n webhook_min_interval_seconds: 29\n", encoding="utf-8")
with pytest.raises(ValidationError, match="below 30"):
load_settings(config_path=path, env_file=tmp_path / "nope.env")
def test_watch_webhook_min_interval_accepts_the_floor(tmp_path):
path = tmp_path / "config.yaml"
path.write_text("watch:\n webhook_min_interval_seconds: 30\n", encoding="utf-8")
s = load_settings(config_path=path, env_file=tmp_path / "nope.env")
assert s.watch.webhook_min_interval_seconds == 30
def test_watch_webhook_url_rejects_a_bad_scheme(tmp_path):
path = tmp_path / "config.yaml"
path.write_text('watch:\n webhook_url: "ftp://example.com/hook"\n', encoding="utf-8")
with pytest.raises(ValidationError, match="http"):
load_settings(config_path=path, env_file=tmp_path / "nope.env")
def test_watch_webhook_on_rejects_an_unknown_entry(tmp_path):
path = tmp_path / "config.yaml"
path.write_text('watch:\n webhook_on: ["crit", "banana"]\n', encoding="utf-8")
with pytest.raises(ValidationError, match="banana"):
load_settings(config_path=path, env_file=tmp_path / "nope.env")
def test_history_trend_default_reports_is_capped(tmp_path): def test_history_trend_default_reports_is_capped(tmp_path):
path = tmp_path / "config.yaml" path = tmp_path / "config.yaml"
path.write_text("history:\n trend_default_reports: 500\n", encoding="utf-8") path.write_text("history:\n trend_default_reports: 500\n", encoding="utf-8")

View file

@ -81,6 +81,27 @@ def test_summarize_cycle_records_a_dead_host_without_inventing_a_number():
assert summary["hosts"]["google-dns"]["loss_pct"] == 100.0 assert summary["hosts"]["google-dns"]["loss_pct"] == 100.0
def test_summarize_cycle_reports_ok_verdict_for_a_healthy_tick():
summary = summarize_cycle(1, "2026-08-08T19:12:00Z", [ping("cloudflare-dns", 12.4)], None)
assert summary["status"] == "ok"
assert summary["score"] == 100
assert summary["finding_ids"] == []
def test_summarize_cycle_reports_a_degraded_verdict_with_finding_ids():
summary = summarize_cycle(1, "2026-08-08T19:12:00Z", [ping("cloudflare-dns", None, loss=100.0)], None)
assert summary["status"] == "crit"
assert summary["score"] < 100
assert any(fid.startswith("latency.unreachable") for fid in summary["finding_ids"])
def test_summarize_cycle_verdict_considers_a_bad_speed_cycle_too():
speed = SpeedResult(method="cloudflare", download_mbps=100.0, upload_mbps=20.0, bufferbloat_down_ms=250.0)
summary = summarize_cycle(1, "2026-08-08T19:12:00Z", [ping("cloudflare-dns", 12.4)], speed)
assert summary["status"] in ("warn", "crit")
assert any(fid.startswith("speed.bufferbloat") for fid in summary["finding_ids"])
def test_session_history_returns_the_series_for_one_host_with_gaps_preserved(): def test_session_history_returns_the_series_for_one_host_with_gaps_preserved():
session = WatchSession(started_at="2026-08-08T19:12:00Z", asn="AS64500") session = WatchSession(started_at="2026-08-08T19:12:00Z", asn="AS64500")
session.add(summarize_cycle(1, "t1", [ping("cloudflare-dns", 12.0)], None)) session.add(summarize_cycle(1, "t1", [ping("cloudflare-dns", 12.0)], None))