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

render TLS, DNS advanced, path diversity, prefix bench, DPI sections in reports

Browse source
This commit is contained in:
Andrew Pimenov 2026-08-09 17:56:07 +03:00
parent fc502f1756
commit 342d92e709
3 changed files with 528 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

202
src/netsleuth/exporter.py
View file

@ -17,8 +17,13 @@ SECTION_ORDER = (
"vpn_assessment",
"bgp",
"reputation",
"dns_advanced",
"latency",
"tls",
"path",
"path_diversity",
"prefix_benchmark",
"dpi_check",
"speed",
)
@ -238,8 +243,13 @@ _SECTION_TITLES = {
"vpn_assessment": "## VPN / proxy assessment",
"bgp": "## ASN & BGP intelligence",
"reputation": "## Reputation",
"dns_advanced": "## DNS: system vs DoH",
"latency": "## Latency",
"tls": "## TLS handshake",
"path": "## Path",
"path_diversity": "## Path diversity (Anycast)",
"prefix_benchmark": "## AS prefix benchmark",
"dpi_check": "## DPI / port check",
"speed": "## Speed",
}
@ -249,8 +259,13 @@ _SECTION_TITLES_RU = {
"vpn_assessment": "## Оценка VPN/прокси",
"bgp": "## ASN и BGP-разведка",
"reputation": "## Репутация",
"dns_advanced": "## DNS: системный резолвер vs DoH",
"latency": "## Задержки",
"tls": "## TLS-рукопожатие",
"path": "## Маршрут",
"path_diversity": "## Диверсификация маршрута (Anycast)",
"prefix_benchmark": "## Бенчмарк префиксов AS",
"dpi_check": "## Проверка DPI / портов",
"speed": "## Скорость",
}
@ -653,6 +668,188 @@ def _speed(report: dict, lang: str = "en") -> list[str]:
return lines + [""]
def _tls(report: dict, lang: str = "en") -> list[str]:
title = _section_title("tls", lang)
note = unavailable(report, "tls", lang)
if note:
return [title, "", note, ""]
results = _data(report, "tls") or []
lines = [title, ""] + _table(
[
_t(lang, "Target", "Цель"),
_t(lang, "Address", "Адрес"),
_t(lang, "TCP RTT (ms)", "TCP RTT (мс)"),
_t(lang, "TLS handshake (ms)", "TLS-рукопожатие (мс)"),
_t(lang, "TTFB (ms)", "TTFB (мс)"),
_t(lang, "Version", "Версия"),
_t(lang, "Cipher", "Шифр"),
],
[
[
r.get("label", ""),
r.get("resolved_ip") or r.get("host", ""),
_num(r.get("tcp_rtt_ms")),
_num(r.get("tls_handshake_ms")),
_num(r.get("ttfb_ms")),
r.get("tls_version") or "—",
r.get("cipher") or "—",
]
for r in results
],
lang,
)
errored = [r for r in results if r.get("error")]
if errored:
lines += [""] + [f"- {r.get('label')}: {r.get('error')}" for r in errored]
lines += [
"",
f"> {_t(lang, 'TLS handshake time is derived by subtracting a separate TCP-only baseline connection from a second, full TLS connection.', 'Время TLS-рукопожатия получено вычитанием отдельного базового TCP-соединения из времени второго, полного TLS-соединения.')}",
]
return lines + [""]
def _dns_advanced(report: dict, lang: str = "en") -> list[str]:
title = _section_title("dns_advanced", lang)
note = unavailable(report, "dns_advanced", lang)
if note:
return [title, "", note, ""]
adv = _data(report, "dns_advanced") or {}
probes = adv.get("probes") or []
lines = [title, ""] + _table(
[
_t(lang, "Resolver", "Резолвер"),
_t(lang, "Kind", "Тип"),
_t(lang, "Name", "Имя"),
_t(lang, "ms", "мс"),
_t(lang, "Answers", "Ответы"),
],
[
[
p.get("name", ""),
p.get("kind", ""),
p.get("query_name", ""),
_num(p.get("elapsed_ms")),
", ".join(p.get("answers") or []) or (p.get("error") or "—"),
]
for p in probes
],
lang,
)
proxy = adv.get("transparent_proxy")
proxy_word = {True: _t(lang, "yes", "да"), False: _t(lang, "no", "нет"), None: "—"}[proxy]
lines += [
"",
f"{_t(lang, 'Transparent DNS proxy', 'Прозрачный DNS-прокси')}: **{proxy_word}** — {adv.get('transparent_proxy_detail') or ''}",
]
note_text = (adv.get("note_ru") or adv.get("note") or "") if lang == "ru" else (adv.get("note") or "")
if note_text:
lines += ["", note_text]
return lines + [""]
def _path_diversity(report: dict, lang: str = "en") -> list[str]:
title = _section_title("path_diversity", lang)
note = unavailable(report, "path_diversity", lang)
if note:
return [title, "", note, ""]
pd = _data(report, "path_diversity") or {}
hops = pd.get("hops") or []
lines = [title, ""] + _table(
[
_t(lang, "Target", "Цель"),
_t(lang, "IP", "IP"),
_t(lang, "IP geo", "Гео IP"),
_t(lang, "Edge colo", "Точка edge"),
_t(lang, "Edge geo", "Гео edge"),
_t(lang, "Edge RTT (ms)", "RTT edge (мс)"),
_t(lang, "Client RTT (ms)", "RTT клиента (мс)"),
],
[
[
h.get("target", ""),
h.get("resolved_ip") or "—",
f"{h.get('ip_city') or '?'}, {h.get('ip_country')}" if h.get("ip_country") else "—",
h.get("edge_colo") or "—",
f"{h.get('edge_city') or '?'}, {h.get('edge_country')}" if h.get("edge_country") else "—",
_num(h.get("edge_rtt_ms")),
_num(h.get("client_rtt_ms")),
]
for h in hops
],
lang,
)
loop_word = _t(lang, "yes", "да") if pd.get("international_loop") else _t(lang, "no", "нет")
lines += ["", f"{_t(lang, 'International routing loop', 'Международная петля маршрута')}: **{loop_word}**"]
note_text = (pd.get("note_ru") or pd.get("note") or "") if lang == "ru" else (pd.get("note") or "")
if note_text:
lines += ["", note_text]
return lines + [""]
def _prefix_benchmark(report: dict, lang: str = "en") -> list[str]:
title = _section_title("prefix_benchmark", lang)
note = unavailable(report, "prefix_benchmark", lang)
if note:
return [title, "", note, ""]
bench = _data(report, "prefix_benchmark") or {}
results = bench.get("results") or []
lines = [title, ""] + _table(
[
_t(lang, "Prefix", "Префикс"),
_t(lang, "Probe IP", "IP пробы"),
_t(lang, "Avg ms", "Сред. мс"),
_t(lang, "Loss", "Потери"),
],
[
[r.get("prefix", ""), r.get("probe_ip") or "—", _num(r.get("avg_ms")), f"{r.get('loss_pct', 0)}%"]
for r in results
],
lang,
)
lines += [
"",
f"{_t(lang, 'Prefixes announced', 'Анонсировано префиксов')}: {bench.get('prefixes_announced', 0)} · "
f"{_t(lang, 'probed', 'проверено')}: {bench.get('prefixes_probed', 0)} · "
f"{_t(lang, 'best', 'лучший')}: {bench.get('best') or '—'} · "
f"{_t(lang, 'worst', 'худший')}: {bench.get('worst') or '—'} · "
f"{_t(lang, 'spread', 'разброс')}: {_num(bench.get('spread_ms'))} {_t(lang, 'ms', 'мс')}",
]
return lines + [""]
_DPI_VERDICT_RU = {
"clean": "чисто",
"partial_filtering": "частичная фильтрация",
"reset_injection": "инъекция RST",
"unreachable": "недоступен",
"unknown": "неизвестно",
}
def _dpi_check(report: dict, lang: str = "en") -> list[str]:
title = _section_title("dpi_check", lang)
note = unavailable(report, "dpi_check", lang)
if note:
return [title, "", note, ""]
dpi = _data(report, "dpi_check") or {}
ports = dpi.get("ports") or []
lines = [title, ""] + _table(
[
_t(lang, "Port", "Порт"),
_t(lang, "State", "Состояние"),
_t(lang, "RTT (ms)", "RTT (мс)"),
_t(lang, "Detail", "Детали"),
],
[[str(p.get("port", "")), p.get("state", ""), _num(p.get("rtt_ms")), p.get("detail") or "—"] for p in ports],
lang,
)
verdict = dpi.get("verdict", "unknown")
verdict_word = verdict if lang == "en" else _DPI_VERDICT_RU.get(verdict, verdict)
rationale = (dpi.get("rationale_ru") or dpi.get("rationale") or "") if lang == "ru" else (dpi.get("rationale") or "")
lines += ["", f"{_t(lang, 'Verdict', 'Вердикт')}: **{verdict_word}** — {rationale}"]
return lines + [""]
def _problems(report: dict, emoji: bool, lang: str = "en") -> list[str]:
found = _findings(report)
heading = _t(lang, "## Problems & recommendations", "## Проблемы и рекомендации")
@ -699,8 +896,13 @@ def render_markdown(
lines += _vpn(report, lang)
lines += _bgp(report, lang)
lines += _reputation(report, lang)
lines += _dns_advanced(report, lang)
lines += _latency(report, lang)
lines += _tls(report, lang)
lines += _path(report, lang)
lines += _path_diversity(report, lang)
lines += _prefix_benchmark(report, lang)
lines += _dpi_check(report, lang)
lines += _speed(report, lang)
lines += _problems(report, emoji, lang)
lines += _diagnostics(report, lang)

116
tests/test_exporter.py
View file

@ -86,8 +86,13 @@ def test_missing_modules_are_rendered_as_skipped_sections_not_dropped():
"vpn_assessment",
"bgp",
"reputation",
"dns_advanced",
"latency",
"tls",
"path",
"path_diversity",
"prefix_benchmark",
"dpi_check",
"speed",
}
@ -266,15 +271,24 @@ def test_format_extensions_covers_the_three_known_formats():
from netsleuth.exporter import SPARK_CHARS, _forced_interface_rows, badge, first_loss_jump, render_markdown, sparkline
from netsleuth.models import (
AdapterLeakResult,
AnycastHop,
BgpIntel,
CfTrace,
DnsAdvanced,
DnsLeak,
DnsblHit,
DpiCheckResult,
InternetDbResult,
IxpPresence,
PathDiversity,
PortProbe,
PrefixBenchmark,
PrefixProbe,
Reputation,
ResolverProbe,
Signal,
TierAttempt,
TlsResult,
TraceHop,
TraceResult,
VpnAssessment,
@ -364,6 +378,22 @@ def full_modules() -> dict[str, ModuleResult]:
rationale="listed on bl.spamcop.net",
),
),
"dns_advanced": ModuleResult(
name="dns_advanced",
status="ok",
data=DnsAdvanced(
probes=[
ResolverProbe(name="system", kind="system", query_name="cloudflare.com", answers=["1.1.1.1"], elapsed_ms=18.0),
ResolverProbe(name="cloudflare", kind="doh", query_name="cloudflare.com", answers=["104.16.132.229"], elapsed_ms=22.0),
],
system_avg_ms=18.0,
doh_avg_ms=22.0,
transparent_proxy=False,
transparent_proxy_detail="No response from the bogus resolver IP (timeout): transparent DNS proxy not detected.",
note="System resolver and DoH agree within normal CDN variance.",
note_ru="Системный резолвер и DoH согласуются в пределах обычной вариации CDN.",
),
),
"latency": ModuleResult(
name="latency",
status="ok",
@ -384,6 +414,23 @@ def full_modules() -> dict[str, ModuleResult]:
],
duration_ms=2400,
),
"tls": ModuleResult(
name="tls",
status="ok",
data=[
TlsResult(
label="cloudflare",
host="cloudflare.com",
port=443,
resolved_ip="104.16.132.229",
tcp_rtt_ms=12.0,
tls_handshake_ms=45.0,
ttfb_ms=60.0,
tls_version="TLSv1.3",
cipher="TLS_AES_256_GCM_SHA384",
)
],
),
"path": ModuleResult(
name="path",
status="ok",
@ -400,6 +447,61 @@ def full_modules() -> dict[str, ModuleResult]:
)
],
),
"path_diversity": ModuleResult(
name="path_diversity",
status="ok",
data=PathDiversity(
client_country="RU",
hops=[
AnycastHop(
target="cloudflare.com",
resolved_ip="104.16.132.229",
edge_colo="FRA",
edge_city="Frankfurt",
edge_country="DE",
edge_rtt_ms=15.0,
client_rtt_ms=45.0,
source="cf_ray",
)
],
international_loop=True,
detour_countries=["DE"],
note="Anycast routed traffic through DE even though the client is in RU.",
note_ru="Anycast завернул трафик через DE, хотя клиент находится в RU.",
),
),
"prefix_benchmark": ModuleResult(
name="prefix_benchmark",
status="ok",
data=PrefixBenchmark(
asn="AS64500",
prefixes_announced=50,
prefixes_probed=2,
method="icmp",
results=[
PrefixProbe(prefix="203.0.113.0/24", probe_ip="203.0.113.1", avg_ms=12.0, loss_pct=0.0, reachable=True),
PrefixProbe(prefix="198.51.100.0/24", probe_ip="198.51.100.1", avg_ms=None, loss_pct=100.0, reachable=False),
],
best="203.0.113.0/24",
worst="203.0.113.0/24",
),
),
"dpi_check": ModuleResult(
name="dpi_check",
status="ok",
data=DpiCheckResult(
target="203.0.113.9",
resolved_ip="203.0.113.9",
consented=True,
ports=[
PortProbe(port=443, state="open", rtt_ms=20.0),
PortProbe(port=8443, state="filtered"),
],
verdict="partial_filtering",
rationale="Some ports responded (open) while others were silently dropped (filtered).",
rationale_ru="Часть портов ответила (открыты), а часть тихо не ответила (отфильтрованы).",
),
),
"speed": ModuleResult(
name="speed",
status="ok",
@ -492,8 +594,13 @@ def test_markdown_has_every_section_from_the_spec_in_order():
"## VPN / proxy assessment",
"## ASN & BGP intelligence",
"## Reputation",
"## DNS: system vs DoH",
"## Latency",
"## TLS handshake",
"## Path",
"## Path diversity (Anycast)",
"## AS prefix benchmark",
"## DPI / port check",
"## Speed",
"## Problems & recommendations",
"## Run diagnostics",
@ -693,8 +800,8 @@ def test_diagnostics_translates_module_warnings_into_russian():
def test_an_all_modules_failed_run_still_renders_every_section():
text = render_markdown(dead_report())
headings = [line for line in text.splitlines() if line.startswith("## ")]
assert len(headings) == 10
assert text.count("Not available") == 8
assert len(headings) == 15
assert text.count("Not available") == len(SECTION_ORDER)
def test_an_all_modules_failed_report_is_still_valid_strict_json():
@ -794,8 +901,13 @@ def test_render_markdown_default_call_is_byte_identical_to_before_the_lang_featu
"## VPN / proxy assessment",
"## ASN & BGP intelligence",
"## Reputation",
"## DNS: system vs DoH",
"## Latency",
"## TLS handshake",
"## Path",
"## Path diversity (Anycast)",
"## AS prefix benchmark",
"## DPI / port check",
"## Speed",
"## Problems & recommendations",
"## Run diagnostics",

212
tests/test_exporter_l7.py Normal file
View file

@ -0,0 +1,212 @@
from __future__ import annotations
import json
import pytest
from netsleuth.exporter import (
SECTION_ORDER,
_dns_advanced,
_dpi_check,
_path_diversity,
_prefix_benchmark,
_section_title,
_tls,
build_report,
dump_json,
)
from netsleuth.models import (
AnycastHop,
DnsAdvanced,
DpiCheckResult,
ModuleResult,
PathDiversity,
PortProbe,
PrefixBenchmark,
PrefixProbe,
ResolverProbe,
TlsResult,
)
def meta() -> dict:
return {
"run_id": "b7f1",
"started_at": "2026-08-08T19:12:00Z",
"finished_at": "2026-08-08T19:13:04Z",
"mode": "auto",
"target": None,
"flags": {},
"host_os": "Windows",
"capabilities": {},
}
def test_every_section_in_order_has_a_title_in_both_languages():
for section in SECTION_ORDER:
assert _section_title(section, "en").startswith("## ")
assert _section_title(section, "ru").startswith("## ")
@pytest.mark.parametrize(
("section", "renderer"),
[
("tls", _tls),
("dns_advanced", _dns_advanced),
("path_diversity", _path_diversity),
("prefix_benchmark", _prefix_benchmark),
("dpi_check", _dpi_check),
],
)
def test_new_l7_section_shows_a_placeholder_when_skipped(section, renderer):
report = build_report(meta(), {section: ModuleResult(name=section, status="skipped")}, [], {})
lines = renderer(report, "en")
assert any("Not available" in line for line in lines)
def test_tls_section_renders_measurements_and_the_handshake_footnote():
report = build_report(
meta(),
{
"tls": ModuleResult(
name="tls",
status="ok",
data=[
TlsResult(
label="cloudflare",
host="cloudflare.com",
resolved_ip="104.16.132.229",
tcp_rtt_ms=12.0,
tls_handshake_ms=45.0,
ttfb_ms=60.0,
tls_version="TLSv1.3",
cipher="TLS_AES_256_GCM_SHA384",
)
],
)
},
[],
{},
)
text = "\n".join(_tls(report, "en"))
assert "104.16.132.229" in text
assert "45" in text
assert "TLSv1.3" in text
assert "subtracting" in text.lower()
def test_tls_section_lists_errored_targets_separately():
report = build_report(
meta(),
{"tls": ModuleResult(name="tls", status="partial", data=[TlsResult(label="dead", host="dead.test", error="timeout")])},
[],
{},
)
text = "\n".join(_tls(report, "en"))
assert "dead" in text
assert "timeout" in text
def test_dns_advanced_section_renders_probes_and_transparent_proxy_verdict():
adv = DnsAdvanced(
probes=[ResolverProbe(name="system", kind="system", query_name="example.com", answers=["1.2.3.4"], elapsed_ms=15.0)],
transparent_proxy=True,
transparent_proxy_detail="A response was received from the bogus resolver IP: transparent DNS proxy detected.",
note="proxy present",
note_ru="прокси обнаружен",
)
report = build_report(meta(), {"dns_advanced": ModuleResult(name="dns_advanced", status="ok", data=adv)}, [], {})
text = "\n".join(_dns_advanced(report, "en"))
assert "example.com" in text
assert "1.2.3.4" in text
assert "Transparent DNS proxy" in text
assert "detected" in text
def test_dns_advanced_section_in_russian_translates_the_note():
adv = DnsAdvanced(transparent_proxy=False, note="clean", note_ru="чисто")
report = build_report(meta(), {"dns_advanced": ModuleResult(name="dns_advanced", status="ok", data=adv)}, [], {})
text = "\n".join(_dns_advanced(report, "ru"))
assert "чисто" in text
def test_path_diversity_section_renders_hops_and_loop_verdict():
pd = PathDiversity(
client_country="RU",
hops=[
AnycastHop(
target="cloudflare.com",
resolved_ip="104.16.132.229",
edge_colo="FRA",
edge_city="Frankfurt",
edge_country="DE",
edge_rtt_ms=10.0,
client_rtt_ms=80.0,
source="cf_ray",
)
],
international_loop=True,
detour_countries=["DE"],
note="Anycast routed traffic through DE.",
note_ru="Anycast завернул трафик через DE.",
)
report = build_report(meta(), {"path_diversity": ModuleResult(name="path_diversity", status="ok", data=pd)}, [], {})
text = "\n".join(_path_diversity(report, "en"))
assert "cloudflare.com" in text
assert "FRA" in text
assert "Frankfurt" in text
assert "International routing loop" in text
def test_prefix_benchmark_section_renders_results_and_summary():
bench = PrefixBenchmark(
asn="AS64500",
prefixes_announced=50,
prefixes_probed=2,
method="icmp",
results=[
PrefixProbe(prefix="203.0.113.0/24", probe_ip="203.0.113.1", avg_ms=12.0, loss_pct=0.0, reachable=True),
PrefixProbe(prefix="198.51.100.0/24", probe_ip="198.51.100.1", avg_ms=None, loss_pct=100.0, reachable=False),
],
best="203.0.113.0/24",
worst="203.0.113.0/24",
spread_ms=None,
)
report = build_report(meta(), {"prefix_benchmark": ModuleResult(name="prefix_benchmark", status="ok", data=bench)}, [], {})
text = "\n".join(_prefix_benchmark(report, "en"))
assert "203.0.113.0/24" in text
assert "198.51.100.0/24" in text
assert "50" in text
def test_dpi_check_section_renders_ports_and_verdict():
dpi = DpiCheckResult(
target="203.0.113.9",
resolved_ip="203.0.113.9",
consented=True,
ports=[PortProbe(port=443, state="open", rtt_ms=20.0), PortProbe(port=8443, state="filtered")],
verdict="partial_filtering",
rationale="Some ports responded while others were dropped.",
rationale_ru="Часть портов ответила, часть тихо не ответила.",
)
report = build_report(meta(), {"dpi_check": ModuleResult(name="dpi_check", status="ok", data=dpi)}, [], {})
text = "\n".join(_dpi_check(report, "en"))
assert "443" in text
assert "open" in text
assert "8443" in text
assert "filtered" in text
assert "Some ports responded" in text
def test_a_report_with_every_new_section_serializes_with_allow_nan_false():
modules = {
"tls": ModuleResult(name="tls", status="ok", data=[TlsResult(label="cf", host="cloudflare.com")]),
"prefix_benchmark": ModuleResult(name="prefix_benchmark", status="ok", data=PrefixBenchmark()),
"dpi_check": ModuleResult(name="dpi_check", status="ok", data=DpiCheckResult()),
"dns_advanced": ModuleResult(name="dns_advanced", status="ok", data=DnsAdvanced()),
"path_diversity": ModuleResult(name="path_diversity", status="ok", data=PathDiversity()),
}
report = build_report(meta(), modules, [], {})
text = dump_json(report)
back = json.loads(text)
assert back["tls"]["data"][0]["host"] == "cloudflare.com"