Horizontal42/netsleuth
1
0
Fork 0
mirror of https://github.com/Horizontal42/netsleuth.git synced 2026-11-11 01:02:10 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge a55267a0d9 into 8c99ac78b7

Browse source
This commit is contained in:
Andrew Pimenov 2026-09-10 11:46:49 +03:00 • committed by GitHub
commit 1dfde5a900
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 91 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
src/netsleuth/alerting.py
View file

@ -2,6 +2,8 @@ from __future__ import annotations
import httpx
from netsleuth.config import Watch
_BAD_STATUSES = {"warn", "crit"}
_GOOD_STATUSES = {"ok", "info"}
@ -11,8 +13,7 @@ def should_fire(
current: str,
last_fired_at: float | None,
now: float,
min_interval_s: float,
fire_on: set[str],
watch: Watch,
) -> bool:
if previous is None or previous == current:
return False
@ -22,14 +23,19 @@ def should_fire(
event = "recovered"
else:
return False
if event not in fire_on:
if event not in set(watch.webhook_on):
return False
if last_fired_at is not None and (now - last_fired_at) < min_interval_s:
if (
last_fired_at is not None
and (now - last_fired_at) < watch.webhook_min_interval_seconds
):
return False
return True
def build_payload(session_meta: dict, cycle_summary: dict, previous: str | None, current: str) -> dict:
def build_payload(
session_meta: dict, cycle_summary: dict, previous: str | None, current: str
) -> dict:
return {
"tool": "netsleuth",
"asn": session_meta.get("asn"),
@ -42,10 +48,12 @@ def build_payload(session_meta: dict, cycle_summary: dict, previous: str | None,
}
async def post_webhook(client: httpx.AsyncClient, url: str, payload: dict, timeout: float) -> bool:
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):
except httpx.HTTPError, OSError:
return False

70
src/netsleuth/watch.py
View file

@ -99,7 +99,9 @@ class WatchSession:
self.cycles.append(summary)
def history(self, label: str) -> list[float | None]:
return [(cycle["hosts"].get(label) or {}).get("avg_ms") for cycle in self.cycles]
return [
(cycle["hosts"].get(label) or {}).get("avg_ms") for cycle in self.cycles
]
def labels(self) -> list[str]:
seen: list[str] = []
@ -137,7 +139,9 @@ def write_session(session: WatchSession, logs_dir: Path) -> Path:
def render_dashboard(session: WatchSession) -> Table:
table = Table(title=f"netsleuth --watch · {session.asn or 'unknown ASN'} · cycle {len(session.cycles)}")
table = Table(
title=f"netsleuth --watch · {session.asn or 'unknown ASN'} · cycle {len(session.cycles)}"
)
for column in ("Host", "Avg ms", "Jitter", "Loss", "Trend"):
table.add_column(column)
latest = session.cycles[-1]["hosts"] if session.cycles else {}
@ -153,7 +157,9 @@ def render_dashboard(session: WatchSession) -> Table:
caption_parts: list[str] = []
if session.cycles:
last = session.cycles[-1]
caption_parts.append(f"verdict: {last.get('status', '?')} ({last.get('score', '?')}/100)")
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)
if speed:
caption_parts.append(
@ -165,7 +171,6 @@ def render_dashboard(session: WatchSession) -> Table:
return table
async def _measure_cycle(
client: httpx.AsyncClient,
settings: Settings,
@ -186,7 +191,10 @@ async def _measure_cycle(
source_ip=bind.ipv4 if bind else None,
)
speed = None
if is_speedtest_cycle(cycle, session.speedtest_every_n_cycles) and not options.quick:
if (
is_speedtest_cycle(cycle, session.speedtest_every_n_cycles)
and not options.quick
):
result = await run_module(
"speed",
_speed_section(client, settings, options, None),
@ -199,7 +207,6 @@ async def _measure_cycle(
async def _maybe_fire_webhook(
client: httpx.AsyncClient,
webhook_url: str | None,
fire_on: set[str],
settings: Settings,
session: WatchSession,
cycle_summary: dict[str, Any],
@ -215,8 +222,7 @@ async def _maybe_fire_webhook(
current_status,
last_fired_at,
time.perf_counter(),
settings.watch.webhook_min_interval_seconds,
fire_on,
settings.watch,
):
payload = build_payload(
{"asn": session.asn, "interface": session.interface},
@ -224,11 +230,14 @@ async def _maybe_fire_webhook(
previous_status,
current_status,
)
if await post_webhook(client, webhook_url, payload, settings.timeouts.http_seconds):
if await post_webhook(
client, webhook_url, payload, settings.timeouts.http_seconds
):
session.alerts_fired += 1
return time.perf_counter()
return last_fired_at
async def run_watch(settings: Settings, options) -> Path:
# Imported here, not at module import time: cli imports watch lazily for --watch,
# and a module-level import back into cli would close the cycle.
@ -247,14 +256,15 @@ async def run_watch(settings: Settings, options) -> Path:
if options.extra_host:
hosts.append(("target-host", options.extra_host))
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]"
f"on a verdict transition into {sorted(set(settings.watch.webhook_on))}.[/yellow]"
)
previous_status: str | None = None
last_fired_at: float | None = None
@ -269,27 +279,49 @@ async def run_watch(settings: Settings, options) -> Path:
try:
geo, _cf, _flags, _raw = await gather_identity(client, settings.providers)
session.asn = geo.asn
except (httpx.HTTPError, OSError):
except httpx.HTTPError, OSError:
session.asn = None
cycle = 1
try:
with Live(render_dashboard(session), console=console, refresh_per_second=settings.watch.dashboard_refresh_hz) as live:
with Live(
render_dashboard(session),
console=console,
refresh_per_second=settings.watch.dashboard_refresh_hz,
) as live:
while True:
began = time.perf_counter()
pings, speed = await _measure_cycle(
client, settings, options, session, cycle, hosts, caps, bind, _speed_section
client,
settings,
options,
session,
cycle,
hosts,
caps,
bind,
_speed_section,
)
cycle_summary = summarize_cycle(
cycle, utc_now_iso(), pings, speed, settings.thresholds
)
cycle_summary = summarize_cycle(cycle, utc_now_iso(), pings, speed, settings.thresholds)
session.add(cycle_summary)
last_fired_at = await _maybe_fire_webhook(
client, webhook_url, fire_on, settings, session, cycle_summary, previous_status, last_fired_at
client,
webhook_url,
settings,
session,
cycle_summary,
previous_status,
last_fired_at,
)
previous_status = cycle_summary["status"]
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
except (KeyboardInterrupt, asyncio.CancelledError):
except KeyboardInterrupt, asyncio.CancelledError:
pass
path = write_session(session, Path(settings.output.logs_dir))

36
tests/test_alerting.py
View file

@ -3,40 +3,50 @@ from __future__ import annotations
import httpx
from netsleuth.alerting import build_payload, post_webhook, should_fire
from netsleuth.config import Watch
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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire(None, "crit", None, 100.0, watch) 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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("crit", "crit", None, 100.0, watch) 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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("ok", "warn", None, 100.0, watch) is False
def test_should_fire_respects_the_minimum_interval():
assert should_fire("ok", "crit", 100.0, 150.0, 300.0, {"crit"}) is False
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("ok", "crit", 100.0, 150.0, watch) 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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("ok", "crit", 100.0, 450.0, watch) 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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("ok", "crit", None, 100.0, watch) 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
watch1 = Watch(webhook_min_interval_seconds=300, webhook_on=["crit"])
assert should_fire("crit", "ok", None, 100.0, watch1) is False
watch2 = Watch(webhook_min_interval_seconds=300, webhook_on=["crit", "recovered"])
assert should_fire("crit", "ok", None, 100.0, watch2) 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
watch = Watch(webhook_min_interval_seconds=300, webhook_on=["recovered"])
assert should_fire("info", "ok", None, 100.0, watch) is False
def test_build_payload_shape_and_json_serializable():
@ -58,12 +68,16 @@ def test_build_payload_shape_and_json_serializable():
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)
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)
ok = await post_webhook(
client, "https://example.com/hook", {"a": 1}, timeout=5.0
)
assert ok is False