zapretdiscordyoutube/zapretgui
Watch
1
23
Fork
You've already forked zapretgui
 0
Code Issues 101 Pull requests Projects Releases 13 Packages Wiki Activity Actions
zapretgui/tests/test_diagnostics_verdict.py
loop-uh 9a8852ad5c
All checks were successful
Zapret source guards / validate (push) Successful in 11s
Details
Скорость: доля повторных передач TCP за загрузку показывает потери на линии 
За загрузку с каждого сервера сравниваются счётчики TCP системы (Windows — GetTcpStatistics, Linux — /proc/net/snmp): сколько сегментов отправлено и сколько из них повторы. Доля повторов видна в блоке «Скорость» отдельным разделом: потерь почти нет, небольшие или заметные; если потери и у российских серверов, виновата ближняя линия, если только у зарубежных — канал за границу. Тест порядка адресов эталонных серверов больше не зависит от того, какой поток успел первым.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
2026-10-20 12:54:21 -04:00

2272 lines
122 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from __future__ import annotations
import re
import socket
import threading
import time
import unittest
from unittest.mock import patch
import diagnostics.engine as engine
from diagnostics import net_access, reach, run_context, sections
from diagnostics.tls_probe import (
KIND_CERT,
KIND_CONNECT,
KIND_OK,
KIND_RESET,
KIND_TIMEOUT,
KIND_TLS,
ProbeResult,
)
from utils.chrome_https import chrome_get
from diagnostics.verdict import (
ADVICE_HOSTS_STALE,
DnsState,
Level,
ReachState,
TargetOutcome,
judge_dns,
judge_reach,
summarize_service,
)
from utils.dns_wire import (
FAILURE_CANCELLED,
FAILURE_TIMEOUT,
STATUS_ERROR,
STATUS_OK,
STATUS_TIMEOUT,
TYPE_AAAA,
DnsQueryResult,
DnsRecord,
)
from utils.socket_cancel import SocketCancel
from utils.windows_dns_query import DNS_STATUS_NAME_ERROR, DnsAnswer
DISCORD_REAL = ("162.159.137.232", "162.159.138.232")
# Адрес прокси разблокирующего DNS (Comss/Xbox DNS и т. п.).
UNBLOCK_PROXY = ("83.220.169.155",)
def _ok(ip: str = DISCORD_REAL[0], **extra) -> ProbeResult:
return ProbeResult(ip=ip, kind=KIND_OK, status=200, tls_version="TLSv1.3", **extra)
class JudgeDnsTests(unittest.TestCase):
def _judge(self, **overrides):
values = dict(
system_ips=DISCORD_REAL,
system_status=0,
reference_ips=DISCORD_REAL,
hosts_ips=(),
check_kind=KIND_OK,
)
values.update(overrides)
return judge_dns(**values)
def test_same_addresses_as_reference_are_ok(self) -> None:
self.assertEqual(self._judge(check_kind="").state, DnsState.OK)
def test_unblocking_dns_proxy_with_real_certificate_is_not_spoofing(self) -> None:
judgement = self._judge(system_ips=UNBLOCK_PROXY, check_kind=KIND_OK)
self.assertEqual(judgement.state, DnsState.OK)
self.assertIn("не подмена", judgement.reason)
def test_foreign_certificate_on_system_address_is_spoofing(self) -> None:
judgement = self._judge(
system_ips=("5.6.7.8",),
check_kind=KIND_CERT,
check_cert_problem="сертификат выдан другому сайту",
)
self.assertEqual(judgement.state, DnsState.SPOOFED)
self.assertIn("другому сайту", judgement.reason)
def test_dpi_reset_on_different_address_is_not_called_spoofing(self) -> None:
judgement = self._judge(system_ips=("5.6.7.8",), check_kind=KIND_RESET)
self.assertEqual(judgement.state, DnsState.UNKNOWN)
def test_stub_and_private_addresses_are_spoofing(self) -> None:
for ip in ("195.82.146.214", "127.0.0.1", "0.0.0.0", "10.10.10.10"):
with self.subTest(ip=ip):
self.assertEqual(self._judge(system_ips=(ip,)).state, DnsState.SPOOFED)
def test_home_network_address_is_a_home_filter_not_the_provider(self) -> None:
"""Роутер с родительским контролем, Pi-hole, AdGuard Home отвечают адресом домашней сети."""
for ip in ("192.168.1.1", "10.0.0.5", "172.16.3.9"):
with self.subTest(ip=ip):
judgement = self._judge(system_ips=(ip,))
self.assertEqual(judgement.state, DnsState.LOCAL)
self.assertIn("это не провайдер", judgement.reason)
def test_vpn_fake_ip_is_local_not_spoofing(self) -> None:
self.assertEqual(self._judge(system_ips=("198.18.0.12",)).state, DnsState.LOCAL)
def test_hosts_file_entry_is_reported_as_local(self) -> None:
judgement = self._judge(hosts_ips=("1.2.3.4",), system_ips=("1.2.3.4",), check_kind=KIND_CERT)
self.assertEqual(judgement.state, DnsState.LOCAL)
self.assertIn("hosts", judgement.reason)
def test_occasional_nxdomain_with_real_addresses_is_spoofing(self) -> None:
# Хватает и одного раза из трёх: исправный DNS про существующий сайт так не отвечает,
# а у пользователя провайдер подсовывал «сайта нет» именно через раз.
judgement = self._judge(check_kind="", nxdomain_count=1, attempts=3)
self.assertEqual(judgement.state, DnsState.SPOOFED)
self.assertIn("1 из 3", judgement.reason)
def test_nxdomain_for_existing_site_is_dns_block(self) -> None:
judgement = self._judge(system_ips=(), system_status=DNS_STATUS_NAME_ERROR, check_kind="")
self.assertEqual(judgement.state, DnsState.SPOOFED)
def test_silent_dns_is_unknown(self) -> None:
judgement = self._judge(system_ips=(), system_status=1460, reference_ips=(), check_kind="")
self.assertEqual(judgement.state, DnsState.UNKNOWN)
class ReachTests(unittest.TestCase):
def test_reach_states(self) -> None:
self.assertEqual(judge_reach(_ok()), ReachState.OK)
self.assertEqual(judge_reach(None), ReachState.NO_ADDRESS)
for kind in (KIND_RESET, KIND_TLS, KIND_TIMEOUT):
with self.subTest(kind=kind):
self.assertEqual(judge_reach(ProbeResult(ip="1.1.1.1", kind=kind)), ReachState.DPI)
self.assertEqual(judge_reach(ProbeResult(ip="1.1.1.1", kind=KIND_CONNECT)), ReachState.IP_BLOCK)
self.assertEqual(judge_reach(ProbeResult(ip="1.1.1.1", kind=KIND_CERT)), ReachState.CERT)
def test_body_cut_at_16kb_is_freeze_but_short_page_is_ok(self) -> None:
self.assertEqual(judge_reach(_ok(body_cut=True, body_size=16_500)), ReachState.FREEZE)
self.assertEqual(judge_reach(_ok(body_cut=True, body_size=3_000)), ReachState.OK)
self.assertEqual(judge_reach(_ok(body_size=16_500)), ReachState.OK)
class ServiceSummaryTests(unittest.TestCase):
def _outcomes(self, main: ReachState, *others: ReachState, dns: DnsState = DnsState.OK):
items = [TargetOutcome("www.youtube.com", "сайт", main, dns, main=True)]
for index, state in enumerate(others):
items.append(TargetOutcome(f"host{index}", f"часть {index}", state, DnsState.OK))
return items
def test_site_that_opens_is_never_red_even_with_dns_block(self) -> None:
verdict = summarize_service(
"YouTube",
self._outcomes(ReachState.OK, ReachState.OK, dns=DnsState.SPOOFED),
zapret_running=True,
)
self.assertEqual(verdict.level, Level.WARN)
self.assertEqual(verdict.headline, "YouTube открывается")
self.assertIn("www.youtube.com", verdict.dns_note)
self.assertTrue(any("DoH" in item for item in verdict.advice))
def test_site_whose_address_the_computer_dns_denies_is_not_called_open(self) -> None:
# Случай 2026-10-13: фильтр отвечает «такого сайта нет» вместо DNS, проверка идёт по адресу из
# шифрованного DNS и сайт открывает, а браузер спрашивает Windows и остаётся ни с чем.
blind = TargetOutcome("youtube.com", "сайт", ReachState.OK, DnsState.SPOOFED, main=True, dns_blind=True)
verdict = summarize_service("youtube.com", [blind], zapret_running=True)
self.assertEqual(verdict.level, Level.WARN)
self.assertIn("сам сайт доступен, но DNS компьютера отвечает, что такого сайта нет", verdict.headline)
self.assertNotIn("открывается", verdict.headline)
self.assertTrue(any("DoH" in item for item in verdict.advice))
def test_everything_open_is_ok(self) -> None:
verdict = summarize_service("Discord", self._outcomes(ReachState.OK, ReachState.OK), zapret_running=True)
self.assertEqual(verdict.level, Level.OK)
self.assertEqual(verdict.advice, ())
def test_main_blocked_by_dpi_with_zapret_running_suggests_strategy(self) -> None:
verdict = summarize_service("Discord", self._outcomes(ReachState.DPI), zapret_running=True)
self.assertEqual(verdict.level, Level.FAIL)
self.assertIn("не обходит", verdict.headline)
self.assertIn("Подбор стратегии", verdict.advice[0])
def test_main_blocked_without_zapret_suggests_start(self) -> None:
verdict = summarize_service("Discord", self._outcomes(ReachState.DPI), zapret_running=False)
self.assertIn("Запустите Zapret", verdict.advice[0])
def test_broken_secondary_part_is_named(self) -> None:
verdict = summarize_service(
"YouTube", self._outcomes(ReachState.OK, ReachState.FREEZE), zapret_running=True
)
self.assertEqual(verdict.level, Level.WARN)
self.assertIn("часть 0", verdict.headline)
class _Done:
"""Уже готовый результат вместо Future: проверка обрыва без пула потоков."""
def __init__(self, value) -> None:
self._value = value
def result(self, timeout=None):
return self._value
class _Net:
"""Фейковая сеть для движка: DNS, эталон и HTTPS по адресу."""
def __init__(self, *, system=DISCORD_REAL, reference=DISCORD_REAL, status=0, https=None, reference_v6=(), voice=(), freeze=(), cause_facts=None):
# Видеосерверы, которые YouTube «выдал» этой сети; пусто — настоящий узел найти не удалось.
self.video_hosts = ("rr1---sn-test.googlevideo.com",)
self.voice = voice
self.freeze = freeze
self.system = system
self.reference = reference
self.reference_v6 = reference_v6
self.status = status
self.https = https or (lambda host, ip: _ok(ip))
self.calls: list[tuple[str, str]] = []
# Что ответил сервер за рубежом и о каких сайтах его спрашивали.
self.outside: dict[str, str] = {}
self.outside_asked: list[str] = []
# Что «нашла» проверка подсетей; сколько раз её звали.
self.subnets = None
self.subnet_checks = 0
# Какие записи спрашивали у эталонного DNS (A, AAAA).
self.doh_types: list[int] = []
# Что «показали» дополнительные пробы: по умолчанию ничего не выяснили.
self.cause_facts = cause_facts
self.refined: list[str] = []
self.bypass_tools: tuple[str, ...] = ()
# Что написано в чужом сертификате (``utils.cert_reader.CertNames``) и у каких адресов его читали.
self.cert_names = None
self.cert_asked: list[tuple[str, str]] = []
# Адреса сайта из файла hosts: по умолчанию записей нет.
self.hosts: tuple[str, ...] = ()
self.dns_place = None
# Что «показал» QUIC: по умолчанию проверку будто сняли — вывода нет.
self.quic_facts = None
self.quic_asked: list[tuple[str, str]] = []
# Что «показал» набор объёма по одному соединению: по умолчанию обрыва нет.
self.volume_facts = None
self.volume_asked: list[str] = []
self.network: dict = {"external_ip": "", "provider": "", "lines": []}
self.telegram: tuple = ()
self.speed: dict | None = None
# Что «показал» замер потока по имени в проверке одного сайта: по умолчанию замера нет.
self.names: dict | None = None
self.crowd: dict | None = None
self.crowd_tools: list | None = None
# TLS 1.2 / TLS 1.3 / HTTP по отдельности: по умолчанию проверку будто сняли.
self.protocol_facts = None
self.registry = engine.registry.Index()
self.habits = None
self.burst = None
self.protocols_asked: list[tuple[str, str]] = []
self.ipv6 = sections.ipv6_check.Ipv6Verdict(sections.ipv6_check.IPV6_ABSENT, "в этой сети его нет")
# Состояние системы читает реестр и службы: в сценариях движка оно задаётся явно.
self.system_items: tuple = ()
def _quic(self, host, ip, **_kwargs):
self.quic_asked.append((host, ip))
if self.quic_facts is not None:
return self.quic_facts(host, ip)
return engine.quic_probe.QuicFacts(host=host, cancelled=True)
def _protocols(self, host, ip, **_kwargs):
self.protocols_asked.append((host, ip))
if self.protocol_facts is not None:
return self.protocol_facts(host, ip)
return engine.protocol_probe.ProtocolFacts(host=host, ip=ip, cancelled=True)
def _volume(self, host, ip, _path, **_kwargs):
self.volume_asked.append(host)
if self.volume_facts is not None:
return self.volume_facts(host, ip)
return engine.volume_probe.VolumeFacts(engine.volume_probe.VolumeRun(engine.volume_probe.RUN_PASSED, 49_500, 4))
def _collect(self, host, result, **_kwargs):
self.refined.append(host)
if self.cause_facts is not None:
return self.cause_facts(host, result)
return engine.block_cause.CauseFacts(host=host, result=result)
def _outside(self, hosts, **_kwargs):
self.outside_asked.extend(hosts)
return {host: self.outside[host] for host in hosts if host in self.outside}
def _subnets(self, _run, _emit):
self.subnet_checks += 1
return self.subnets
def _doh(self, _run, _host, record_type=engine.DNS_TYPE_A):
self.doh_types.append(record_type)
return True, self.reference_v6 if record_type == TYPE_AAAA else self.reference
def patches(self):
def _https_get(host, ip, *_args, **_kwargs):
self.calls.append((host, ip))
return self.https(host, ip)
return (
# Паузы между стартами сайтов — против узора на линии; в поддельной сети они только тормозят тесты.
patch.object(run_context, "SITE_START_GAP_S", (0.0, 0.0)),
patch.object(
net_access,
"query_ipv4",
side_effect=lambda host, **_kw: (
self.system(host) if callable(self.system) else DnsAnswer(ips=self.system, status=self.status)
),
),
patch.object(
net_access,
"doh_lookup",
side_effect=self._doh,
),
patch.object(net_access, "chrome_get", side_effect=_https_get),
# Пауза перед повтором нужна настоящей сети; сценариям она только добавляет секунды.
patch.object(reach, "RETRY_PAUSE_S", 0.0),
# Уточнение причины ходит в сеть само: в сценариях движка оно подменено.
patch.object(engine.block_cause, "collect", side_effect=self._collect),
patch.object(engine.quic_probe, "collect", side_effect=self._quic),
patch.object(engine.volume_probe, "collect", side_effect=self._volume),
patch.object(engine.protocol_probe, "collect", side_effect=self._protocols),
patch.object(sections, "check_ipv6", side_effect=lambda _run: self.ipv6),
# «Ваша сеть» и дата-центры Telegram ходят в сеть сами: в сценариях движка их нет.
patch.object(sections, "check_network", side_effect=lambda _run, _tools: self.network),
patch.object(sections, "check_speed", side_effect=lambda _run, *_sites: self.speed),
# Замер потока по имени в проверке одного сайта качает файл с постороннего сервера.
patch.object(sections, "check_name_speed", side_effect=lambda *_a, **_k: self.names),
# Пачка одновременных соединений — настоящие приветствия в сеть.
patch.object(sections, "check_crowd", side_effect=self._crowd),
patch.object(engine.telegram_check, "check_telegram", side_effect=lambda *_a, **_k: self.telegram),
patch.object(sections, "check_system", side_effect=lambda _run, _services, _zapret=None, **_kwargs: self.system_items),
# Подсети — десятки настоящих соединений: в сценариях движка их нет.
patch.object(sections, "check_subnets", side_effect=self._subnets),
# «Жив ли сайт снаружи» — запрос к серверу проекта: в сценариях его нет.
patch.object(sections.outside_check, "ask", side_effect=self._outside),
patch.object(net_access, "hosts_file_ipv4", side_effect=lambda _host: self.hosts),
# Чужой сертификат читается отдельным соединением — в сценариях движка сети нет.
patch.object(engine.cert_owner, "collect", side_effect=self._cert),
patch.object(net_access, "system_dns_servers", return_value=("83.220.169.155",)),
patch.object(sections, "zapret_status", return_value=(True, "✅ Zapret запущен")),
# Список реестра РКН качается из сети и лежит на диске: в сценариях движка его нет.
patch.object(sections, "start_registry", return_value=lambda _seconds, _cancel: self.registry),
# Пробы «как работает фильтр» шлют свои приветствия в сеть.
patch.object(sections, "check_filter_habits", side_effect=lambda _run, _collected, **_kwargs: self.habits),
# Место перехвата DNS — настоящие пакеты в сеть: в сценариях движка его нет.
patch.object(sections, "_dns_place", side_effect=lambda _run, _trace: self.dns_place),
# Серии пакетов UDP уходят на настоящие серверы.
patch.object(sections, "check_udp_burst", side_effect=lambda _run: self.burst),
patch.object(engine, "running_bypass_tools", return_value=self.bypass_tools),
patch.object(engine, "_discover_googlevideo", side_effect=lambda _run: (self.video_hosts, "" if self.video_hosts else "видеосервер не проверялся")),
patch.object(engine.video_nodes, "owner_note", return_value=""),
# Звонки и обрыв 16 КБ проверяются в любом режиме — без сети в тестах.
patch("diagnostics.voice_check.check_voice", return_value=self.voice),
patch("diagnostics.freeze_check.check_freeze", return_value=self.freeze),
)
def _crowd(self, _run, _collected, _services, *, tools=()):
self.crowd_tools = list(tools)
return self.crowd
def _cert(self, host, ip, **_kwargs):
self.cert_asked.append((host, ip))
return self.cert_names
def run(self, fn, *args, **kwargs):
from contextlib import ExitStack
with ExitStack() as stack:
for item in self.patches():
stack.enter_context(item)
return fn(*args, **kwargs)
class ReferenceResolverTests(unittest.TestCase):
"""Эталон собирается с нескольких серверов; закрытый провайдером сервер виден в отчёте."""
@staticmethod
def _answer(rtype, *ips) -> DnsQueryResult:
records = tuple(DnsRecord(name="x", rtype=rtype, ttl=60, value=ip) for ip in ips)
return DnsQueryResult(status=STATUS_OK, rcode=0, records=records, elapsed_ms=10.0)
def _run(self, doh, fn, *args, **kwargs):
from contextlib import ExitStack
net = _Net()
with ExitStack() as stack:
for item in net.patches():
if getattr(item, "attribute", "") != "doh_lookup":
stack.enter_context(item)
stack.enter_context(patch.object(net_access, "query_doh", doh))
return fn(*args, **kwargs)
def _doh(self, *, silent=(), cancelled=()):
def doh(server, _host, rtype, **_kwargs):
if server in silent:
return DnsQueryResult(status=STATUS_TIMEOUT, failure=FAILURE_TIMEOUT)
if server in cancelled:
return DnsQueryResult(status=STATUS_ERROR, failure=FAILURE_CANCELLED)
return self._answer(rtype, *DISCORD_REAL) if rtype == engine.DNS_TYPE_A else self._answer(rtype)
return doh
def test_blocked_reference_server_is_named_with_reason_and_dns_button(self) -> None:
lines: list[str] = []
result = self._run(self._doh(silent=("8.8.8.8",)), engine.run_blockcheck, "main", emit=lines.append)
state = {item["address"]: item for item in result["reference"]}
self.assertFalse(state["8.8.8.8"]["ok"])
self.assertEqual(state["8.8.8.8"]["reason"], "сервер молчит")
self.assertTrue(state["1.1.1.1"]["ok"])
problem = next(item for item in result["problems"] if "8.8.8.8" in item["text"])
self.assertEqual((problem["level"], problem["action"]), ("warn", "dns"))
self.assertIn("Шифрованный DNS Google (8.8.8.8) недоступен: сервер молчит", "\n".join(lines))
# Остальные серверы ответили — эталон есть, сайты проверены как обычно.
self.assertIn("Discord", result["working"])
def test_dns_tab_also_names_blocked_reference_server(self) -> None:
lines: list[str] = []
result = self._run(self._doh(silent=("8.8.8.8",)), engine.run_dns_check, emit=lines.append)
self.assertIn("Шифрованный DNS Google (8.8.8.8) недоступен", "\n".join(lines))
self.assertFalse(result["summary"]["dns_poisoning_detected"])
self.assertFalse({item["address"]: item["ok"] for item in result["reference"]}["8.8.8.8"])
def test_when_every_reference_is_silent_no_single_server_is_blamed(self) -> None:
"""Молчат все — это не «провайдер закрыл Google», а нет связи вовсе."""
everyone = tuple(resolver.address for resolver in engine.REFERENCE_RESOLVERS)
lines: list[str] = []
result = self._run(self._doh(silent=everyone), engine.run_blockcheck, "main", emit=lines.append)
self.assertTrue(all(not item["ok"] for item in result["reference"]))
self.assertFalse([item for item in result["problems"] if "Шифрованный DNS" in item["text"]])
self.assertIn("эталон: недоступен", "\n".join(lines))
def test_query_stopped_by_time_limit_says_nothing_about_the_server(self) -> None:
result = self._run(self._doh(cancelled=("8.8.8.8",)), engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertNotIn("8.8.8.8", {item["address"] for item in result["reference"]})
self.assertFalse([item for item in result["problems"] if "Шифрованный DNS" in item["text"]])
def test_silent_reference_server_does_not_hold_every_site(self) -> None:
# Живой случай: один эталонный сервер закрыт провайдером и молчит весь срок.
# Раньше каждый сайт ждал его пять секунд.
release = threading.Event()
def doh(server, _host, rtype, **_kwargs):
if server == "8.8.8.8":
release.wait(5)
return self._answer(rtype, "8.8.8.80")
return self._answer(rtype, "1.1.1.10")
run = engine._Run(None, workers=8)
try:
with patch.object(net_access, "query_doh", doh), patch.object(net_access, "REFERENCE_GRACE_S", 0.1):
started = time.monotonic()
answered, ips = net_access.doh_lookup(run, "example.com")
waited = time.monotonic() - started
self.assertNotIn("8.8.8.8", {item["address"] for item in run.reference_report()})
release.set()
deadline = time.monotonic() + 2
while time.monotonic() < deadline and "8.8.8.8" not in {item["address"] for item in run.reference_report()}:
time.sleep(0.01)
# Опоздавший ответ всё равно попал в отчёт о серверах.
self.assertIn("8.8.8.8", {item["address"] for item in run.reference_report()})
finally:
release.set()
run.close()
self.assertTrue(answered)
self.assertEqual(ips, ("1.1.1.10",))
self.assertLess(waited, 1.5)
def test_answers_of_all_reference_servers_are_merged(self) -> None:
by_server = {"1.1.1.1": ("1.1.1.10",), "8.8.8.8": ("8.8.8.80", "1.1.1.10")}
def doh(server, _host, rtype, **_kwargs):
return self._answer(rtype, *by_server.get(server, ()))
run = engine._Run(None, workers=8)
try:
with patch.object(net_access, "query_doh", doh):
answered, ips = net_access.doh_lookup(run, "example.com")
finally:
run.close()
self.assertTrue(answered)
# Серверы отвечают в потоках: порядок адресов зависит от того, кто успел первым.
self.assertEqual(sorted(ips), ["1.1.1.10", "8.8.8.80"])
class BlockCauseInReportTests(unittest.TestCase):
"""Сайт не открылся: в отчёте сказано, режут по имени или по адресу."""
@staticmethod
def _reset_on_hello(host, ip):
return ProbeResult(ip=ip, kind=KIND_RESET, stage="tls")
def test_block_by_name_is_shown_in_report_and_first_in_advice(self) -> None:
def facts(host, result):
return engine.block_cause.CauseFacts(
host=host,
result=result,
neutral=engine.block_cause.HelloResult(engine.block_cause.HELLO_OK),
nameless=engine.block_cause.HelloResult(engine.block_cause.HELLO_RESET),
)
lines: list[str] = []
net = _Net(https=self._reset_on_hello, cause_facts=facts)
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
self.assertIn(" ❌ Как блокируют: Блокировка по имени сайта: с именем discord.com соединение обрывается", "\n".join(lines))
discord = next(item for item in result["services"] if item["key"] == "discord")
main = next(item for item in discord["targets"] if item["main"])
self.assertEqual(main["cause"], "by_name")
self.assertTrue(main["cause_text"].startswith("Блокировка по имени сайта"))
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertTrue(problem["advice"][0].startswith("Блокировка по имени сайта"))
# От блокировки по имени стратегия помогает: кнопка подбора остаётся.
self.assertEqual(problem["action"], "strategy")
def test_working_site_is_not_probed_again(self) -> None:
net = _Net()
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertEqual(net.refined, [])
self.assertTrue(all(target["cause"] == "" for item in result["services"] for target in item["targets"]))
def test_unclear_probes_add_nothing_to_report(self) -> None:
lines: list[str] = []
net = _Net(https=self._reset_on_hello)
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
self.assertIn("discord.com", net.refined)
self.assertNotIn("🔎", "\n".join(lines))
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertIn("Подбор стратегии", problem["advice"][0])
def test_other_bypass_tools_are_named_before_results(self) -> None:
lines: list[str] = []
net = _Net()
net.bypass_tools = ("Xray", "Cloudflare WARP")
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
note = next(line for line in lines if "Xray" in line)
# Про адаптеры здесь узнать нельзя: VPN остаётся под подозрением, прокси — нет.
self.assertIn("На дороге проверки стоит: Cloudflare WARP", note)
self.assertIn("Работает прокси: Xray", note)
self.assertLess(lines.index(note), next(i for i, line in enumerate(lines) if line.startswith("━━━━━━━━ Discord")))
self.assertEqual(result["other_bypass_tools"], ["Xray", "Cloudflare WARP"])
class QuicInReportTests(unittest.TestCase):
"""QUIC проверяется по тому же адресу, что и сайт, и не путается с блокировкой сайта."""
@staticmethod
def _facts(blocked=(), silent=()):
def facts(host, _ip):
if host in blocked:
return engine.quic_probe.QuicFacts(host=host, neutral_ms=40.0, nameless_ms=41.0)
if host in silent:
return engine.quic_probe.QuicFacts(host=host)
return engine.quic_probe.QuicFacts(host=host, real_ms=12.0, neutral_ms=13.0, nameless_ms=13.0)
return facts
def test_working_quic_is_shown_and_is_not_a_problem(self) -> None:
lines: list[str] = []
net = _Net()
net.quic_facts = self._facts()
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
self.assertIn(" ✅ QUIC (UDP 443): отвечает за 12 мс", lines)
self.assertFalse([item for item in result["problems"] if "QUIC" in item["text"]])
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertEqual({target["quic"] for target in discord["targets"]}, {"ok"})
# Пакеты идут на тот же адрес, к которому шёл основной запрос.
self.assertIn(("discord.com", DISCORD_REAL[0]), net.quic_asked)
def test_quic_blocked_for_open_site_is_one_warning_naming_services(self) -> None:
lines: list[str] = []
net = _Net()
net.quic_facts = self._facts(blocked={"www.youtube.com", "i.ytimg.com", "discord.com"})
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
text = "\n".join(lines)
self.assertIn(" ⚠️ QUIC (UDP 443): блокируется по имени сайта: пакет с именем discord.com пропадает", text)
problems = [item for item in result["problems"] if "QUIC" in item["text"]]
self.assertEqual(len(problems), 1)
self.assertEqual(problems[0]["level"], "warn")
self.assertIn("для: Discord, YouTube.", problems[0]["text"])
self.assertIn("UDP 443", problems[0]["advice"][0])
# Сами сайты открываются: «не открывается» про них не пишется.
self.assertEqual(sorted(result["working"]), ["Discord", "YouTube"])
def test_server_without_quic_is_a_note_not_a_problem(self) -> None:
lines: list[str] = []
net = _Net()
net.quic_facts = self._facts(silent={"discord.com"})
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
note = next(line for line in lines if "QUIC" in line and "не поддерживает" in line)
self.assertTrue(note.startswith(" ℹ️ "))
self.assertFalse([item for item in result["problems"] if "QUIC" in item["text"]])
def test_blocked_site_gets_site_problem_not_extra_quic_one(self) -> None:
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET, stage="tls"))
net.quic_facts = self._facts(blocked={"discord.com", "gateway.discord.gg", "cdn.discordapp.com"})
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertFalse([item for item in result["problems"] if item["text"].startswith("QUIC")])
self.assertTrue([item for item in result["problems"] if item["target"] == "discord.com"])
def test_dns_tab_does_not_send_quic_packets(self) -> None:
net = _Net()
net.quic_facts = self._facts()
net.run(engine.run_dns_check, emit=lambda _line: None)
self.assertEqual(net.quic_asked, [])
class Ipv6InReportTests(unittest.TestCase):
def _run(self, code: str, text: str):
lines: list[str] = []
net = _Net()
net.ipv6 = sections.ipv6_check.Ipv6Verdict(code, text)
return net.run(engine.run_blockcheck, "main", emit=lines.append), lines
def test_broken_ipv6_is_a_warning_with_advice(self) -> None:
result, lines = self._run(sections.ipv6_check.IPV6_BROKEN, "настроен, но не работает")
self.assertIn("⚠️ IPv6 настроен, но не работает", lines)
problem = next(item for item in result["problems"] if item["text"].startswith("IPv6"))
self.assertEqual(problem["level"], "warn")
self.assertIn("с задержкой", problem["advice"][0])
self.assertEqual(result["ipv6"], {"state": "broken", "text": "настроен, но не работает"})
def test_absent_and_working_ipv6_are_only_lines_in_report(self) -> None:
for code, icon in ((sections.ipv6_check.IPV6_ABSENT, "❔"), (sections.ipv6_check.IPV6_OK, "✅")):
with self.subTest(code=code):
result, lines = self._run(code, "текст")
self.assertIn(f" {icon} IPv6 текст", lines)
self.assertFalse([item for item in result["problems"] if item["text"].startswith("IPv6")])
def test_ipv6_section_comes_after_sites_and_before_summary(self) -> None:
_result, lines = self._run(sections.ipv6_check.IPV6_OK, "работает")
section = lines.index("━━━━━━━━ IPv6 ━━━━━━━━")
self.assertGreater(section, next(i for i, line in enumerate(lines) if line.startswith("━━━━━━━━ YouTube")))
self.assertLess(section, lines.index("━━━━━━━━ 📊 Итог ━━━━━━━━"))
class FreezeUploadWiringTests(unittest.TestCase):
"""Отправка проверяется на том же адресе, по которому шла загрузка."""
def _run(self):
run = engine._Run(None, workers=4)
self.addCleanup(run.close)
return run
def test_address_is_found_once_and_reused_for_upload(self) -> None:
run = self._run()
asked: list = []
def collect(host, ip, path, **_kwargs):
asked.append((host, ip, path))
return sections.upload_probe.UploadFacts(sections.upload_probe.PostResult(sections.upload_probe.POST_STALLED))
with (
patch.object(net_access, "hosts_file_ipv4", return_value=("203.0.113.9",)) as hosts,
patch.object(net_access, "chrome_get", return_value=_ok("203.0.113.9")) as get,
patch.object(sections.upload_probe, "collect", collect),
):
sections.download(run, "cdn.example", "/file.bin")
verdict = sections.upload(run, "cdn.example", "/file.bin")
self.assertEqual(hosts.call_count, 1)
self.assertEqual(get.call_args.args[:2], ("cdn.example", "203.0.113.9"))
self.assertEqual(asked, [("cdn.example", "203.0.113.9", "/file.bin")])
self.assertEqual(verdict.code, sections.upload_probe.UPLOAD_UNKNOWN)
def test_no_address_or_stopped_run_means_no_upload_probe(self) -> None:
run = self._run()
with (
patch.object(net_access, "hosts_file_ipv4", return_value=()),
patch.object(net_access, "query_ipv4", return_value=DnsAnswer()),
patch.object(net_access, "doh_lookup", return_value=(False, ())),
patch.object(sections.upload_probe, "collect") as collect,
):
self.assertIsNone(sections.upload(run, "nowhere.example", "/"))
collect.assert_not_called()
stopped = engine._Run(lambda: True, workers=4)
self.addCleanup(stopped.close)
with (
patch.object(net_access, "hosts_file_ipv4", return_value=("203.0.113.9",)),
patch.object(sections.upload_probe, "collect") as collect,
):
self.assertIsNone(sections.upload(stopped, "cdn.example", "/"))
collect.assert_not_called()
class ForeignCertificateTests(unittest.TestCase):
"""Чужой сертификат: отчёт называет, кто ответил вместо сайта, и что с этим делать."""
HOSTS_IP = "203.0.113.9"
@staticmethod
def _names(**fields):
from utils.cert_reader import CertNames
return CertNames(**fields)
def _discord(self, result) -> tuple[dict, dict]:
service = next(item for item in result["services"] if item["label"] == "Discord")
return service, next(item for item in service["targets"] if item["host"] == "discord.com")
def test_stale_hosts_record_is_a_problem_even_though_the_real_address_opens(self) -> None:
# Живой случай (ChatGPT у пользователя): в hosts записан сервер-посредник, которого больше нет.
# Проверка открыла сайт по настоящему адресу, а браузер пойдёт по записи и упрётся в чужой сервер.
def https(host, ip):
if ip == self.HOSTS_IP:
return ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан другому сайту")
return _ok(ip)
net = _Net(https=https)
net.hosts = (self.HOSTS_IP,)
net.cert_names = self._names(subject="other.example", issuer="R11", issuer_org="Let's Encrypt", names=("other.example",))
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
service, target = self._discord(result)
self.assertTrue(target["hosts_stale"])
self.assertEqual(service["level"], "warn")
self.assertEqual(target["cert"]["code"], "hosts")
self.assertIn("other.example", target["cert"]["text"])
self.assertIn(("discord.com", self.HOSTS_IP), net.cert_asked)
problem = next(item for item in result["problems"] if "Discord" in item["text"])
self.assertEqual((problem["level"], problem["action"]), ("warn", "hosts"))
self.assertIn("в файле hosts для него записан нерабочий адрес", problem["text"])
self.assertTrue(any("чужой сервер" in line for line in problem["advice"]))
self.assertNotIn("Discord", result["working"])
def test_antivirus_inspecting_https_is_named_in_the_problem(self) -> None:
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан неизвестным центром"))
net.cert_names = self._names(subject="discord.com", issuer="ESET SSL Filter CA", names=("discord.com",))
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
_service, target = self._discord(result)
self.assertEqual(target["cert"]["code"], "antivirus")
self.assertIn("ESET", target["text"])
problem = next(item for item in result["problems"] if "Discord" in item["text"])
self.assertTrue(any("ESET" in line for line in problem["advice"]))
self.assertEqual(problem["action"], "")
def test_site_with_valid_certificate_has_no_certificate_note(self) -> None:
net = _Net()
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
self.assertIsNone(self._discord(result)[1]["cert"])
self.assertEqual(net.cert_asked, [])
class SiteWordsComeFromTheReportTests(unittest.TestCase):
"""Слово итога, дороги и метки сайта лежат в отчёте готовыми: экран их только показывает."""
@staticmethod
def _protocols(tls12, tls13):
bc = engine.block_cause
return lambda host, ip: engine.protocol_probe.ProtocolFacts(
host=host, ip=ip, tls12=bc.HelloResult(tls12, ms=30.0), tls13=bc.HelloResult(tls13, ms=30.0)
)
@staticmethod
def _discord(result) -> dict:
return next(item for item in result["services"] if item["key"] == "discord")
def test_open_site_has_status_and_its_roads_in_fixed_order(self) -> None:
# Открытому сайту хватает основной проверки «как Chrome»: простые приветствия к нему не шлются.
net = _Net()
net.protocol_facts = lambda host, ip: self.fail("пробы протоколов у открытого сайта")
discord = self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))
self.assertEqual(discord["status"], "Открывается")
self.assertEqual([road["key"] for road in discord["roads"]], ["browser", "quic", "dns"])
self.assertEqual([road["label"] for road in discord["roads"]], ["Chrome", "QUIC", "DNS"])
for road in discord["roads"]:
self.assertTrue(road["word"] and road["text"] and road["hint"], road)
self.assertIn(road["state"], ("ok", "warn", "fail", "info", "unknown"))
dns = next(road for road in discord["roads"] if road["key"] == "dns")
self.assertEqual((dns["state"], dns["word"]), ("ok", "честный"))
def test_closed_site_gets_every_road(self) -> None:
bc = engine.block_cause
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "discord.com" else _ok(ip))
net.protocol_facts = self._protocols(bc.HELLO_RESET, bc.HELLO_RESET)
discord = self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))
self.assertEqual([road["key"] for road in discord["roads"]][:3], ["tls12", "tls13", "browser"])
def test_site_cut_only_for_chrome_is_a_fingerprint_block_in_the_report(self) -> None:
# Основная проверка «как Chrome» оборвана, а простое приветствие к тому же адресу проходит.
bc = engine.block_cause
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "discord.com" else _ok(ip))
net.protocol_facts = self._protocols(bc.HELLO_OK, bc.HELLO_OK)
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
discord = self._discord(result)
self.assertEqual((discord["level"], discord["kind"]), ("fail", "fingerprint"))
self.assertIn("похожие на браузер", discord["headline"])
self.assertNotIn("Discord", result["working"])
# Простое приветствие тоже рвут — обычная блокировка, не «почерк».
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "discord.com" else _ok(ip))
net.protocol_facts = self._protocols(bc.HELLO_RESET, bc.HELLO_RESET)
self.assertNotEqual(self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))["kind"], "fingerprint")
def test_status_names_the_cause_that_is_not_the_provider(self) -> None:
def https(host, ip):
if ip == "203.0.113.9":
return ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан другому сайту")
return _ok(ip)
from utils.cert_reader import CertNames
net = _Net(https=https)
net.hosts = ("203.0.113.9",)
net.cert_names = CertNames(subject="other.example", issuer="R11", names=("other.example",))
discord = self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))
self.assertEqual(discord["status"], "Адрес из hosts ведёт не туда")
self.assertIn({"key": "hosts", "text": "запись в hosts устарела", "state": "warn"}, discord["tags"])
def test_blocked_site_status_is_the_kind_of_block(self) -> None:
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET))
discord = self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))
self.assertEqual(discord["level"], "fail")
self.assertNotIn(discord["status"], ("Открывается", "Есть проблемы", ""))
def test_way_of_blocking_is_a_ready_tag(self) -> None:
# Раньше слово «блокировка по имени» для шапки и подсказок собирал экран по коду причины.
bc = engine.block_cause
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET, stage="tls"))
net.cause_facts = lambda host, result: bc.CauseFacts(
host=host,
result=result,
neutral=bc.HelloResult(bc.HELLO_OK),
nameless=bc.HelloResult(bc.HELLO_RESET),
real_again=bc.HelloResult(bc.HELLO_RESET),
)
discord = self._discord(net.run(engine.run_blockcheck, "full", emit=lambda _line: None))
causes = [tag for tag in discord["tags"] if tag["key"] == "cause"]
self.assertEqual({target["cause"] for target in discord["targets"]}, {"by_name"})
self.assertEqual(causes, [{"key": "cause", "text": "блокировка по имени", "state": "fail"}])
def test_single_site_check_carries_the_same_words_and_the_registry_tag(self) -> None:
# «Проверка домена» показывает ту же карточку: слова и метка реестра должны быть и там.
net = _Net()
net.registry = engine.registry.Index(hosts=engine.registry.build_hosts(["example.org"])[0], updated=1000.0)
net.protocol_facts = self._protocols(engine.block_cause.HELLO_OK, engine.block_cause.HELLO_OK)
site = net.run(engine.check_site, "example.org")
self.assertEqual(site["status"], "Открывается")
self.assertTrue(site["roads"])
self.assertIn({"key": "registry", "text": "в реестре РКН", "state": "info"}, site["tags"])
def test_single_site_check_says_when_the_computer_dns_denies_the_site(self) -> None:
net = _Net(system=(), status=DNS_STATUS_NAME_ERROR, reference=("142.251.157.4",))
site = net.run(engine.check_site, "example.org")
self.assertEqual(site["level"], "warn")
self.assertIn("DNS компьютера отвечает, что такого сайта нет", site["headline"])
def test_single_site_check_covers_the_other_addresses_of_a_known_service(self) -> None:
# «youtube.com открывается» мало что значит, когда не отвечают видеосерверы.
from diagnostics.services import related_targets, site_service
self.assertEqual(
[(target.host, target.purpose, target.main) for target in site_service("youtube.com").targets],
[
("youtube.com", "сайт", True),
("www.youtube.com", "основной адрес", False),
("i.ytimg.com", "превью видео", False),
("youtubei.googleapis.com", "служба плеера", False),
(engine.GOOGLEVIDEO_FALLBACK_HOST, "видео", False),
],
)
self.assertTrue(site_service("m.youtube.com").targets[-1].discover_googlevideo)
self.assertEqual([target.host for target in related_targets("gateway.discord.gg")], ["discord.com", "cdn.discordapp.com"])
self.assertEqual(
[target.host for target in related_targets("www.youtube.com")],
["i.ytimg.com", "youtubei.googleapis.com", engine.GOOGLEVIDEO_FALLBACK_HOST],
)
# Чужой сайт и сайт из одного адреса остаются сами по себе.
self.assertEqual(related_targets("example.org"), ())
self.assertEqual(related_targets("notyoutube.com"), ())
self.assertEqual(related_targets("x.com"), ())
net = _Net()
site = net.run(engine.check_site, "youtube.com")
self.assertEqual([target["host"] for target in site["targets"]][:3], ["youtube.com", "www.youtube.com", "i.ytimg.com"])
self.assertEqual(len(site["targets"]), 5)
def test_single_site_check_measures_the_stream_by_name_and_does_not_call_a_cut_stream_open(self) -> None:
# Соединение есть, а поток с именем сайта обрывают: «открывается» было бы неправдой.
names = {
"level": "warn",
"headline": "Обрывают поток по имени сайта: example.org",
"server": "mirror.yandex.ru",
"control": "90 Мбит/с",
"items": [{"name": "example.org", "host": "example.org", "kbps": None, "state": "warn", "text": "поток дважды оборвался"}],
}
net = _Net()
net.names = names
site = net.run(engine.check_site, "example.org")
self.assertEqual(site["names"], names)
self.assertEqual((site["level"], site["status"]), ("warn", "Поток режут"))
self.assertIn("открывается, но поток данных режут по имени: example.org", site["headline"])
self.assertTrue(any("Подбор стратегии" in item or "Запустите Zapret" in item for item in site["advice"]))
calm = dict(names, level="ok", items=[dict(names["items"][0], state="ok")])
net.names = calm
site = net.run(engine.check_site, "example.org")
self.assertEqual((site["level"], site["headline"]), ("ok", "example.org открывается"))
self.assertEqual(site["names"], calm)
def test_control_and_registry_tags(self) -> None:
net = _Net()
net.registry = engine.registry.Index(hosts=engine.registry.build_hosts(["discord.com"])[0], updated=1000.0)
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
self.assertIn({"key": "registry", "text": "в реестре РКН", "state": "info"}, self._discord(result)["tags"])
control = next(item for item in result["services"] if item["control"])
self.assertIn({"key": "control", "text": "контрольный", "state": "info"}, control["tags"])
self.assertNotIn("registry", [tag["key"] for tag in control["tags"]])
class UnstableSiteTests(unittest.TestCase):
"""Третий исход: сайт не «открывается» и не «заблокирован», а открывается через раз."""
@staticmethod
def _discord(result) -> tuple[dict, dict]:
service = next(item for item in result["services"] if item["key"] == "discord")
return service, service["targets"][0]
def test_failure_success_failure_is_reported_as_every_other_time(self) -> None:
seen: dict[str, int] = {}
def _https(host, ip):
seen[host] = seen.get(host, 0) + 1
# Залп — сбой (три попытки), лёгкий повтор и полная перепроверка — успех, третья попытка — снова сбой.
if host == "discord.com" and (seen[host] <= 3 or seen[host] >= 6):
return ProbeResult(ip=ip, kind=KIND_CONNECT, connect_fail="timeout")
return _ok(ip)
net = _Net(https=_https)
with patch.object(reach, "RETRY_PAUSE_S", 0.0):
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
service, target = self._discord(result)
self.assertEqual((target["rechecked"], target["ok"]), ("unstable", True))
self.assertIn("из трёх попыток подряд прошла одна", target["unstable"])
self.assertEqual((service["level"], service["headline"]), ("warn", "Discord открывается через раз"))
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertEqual((problem["level"], problem["action"]), ("warn", "strategy"))
# Вид исхода назван в самом отчёте: экран не должен его угадывать.
self.assertEqual((service["kind"], problem["kind"]), ("unstable", "unstable"))
self.assertTrue(any("трёх попыток подряд прошла одна" in line for line in problem["advice"]))
self.assertNotIn("Discord", result["working"])
def test_cut_on_one_address_of_two_is_every_other_time_not_a_full_block(self) -> None:
# Живое наблюдение из жалоб: обрыв на 16 КБ бывает на одном адресе сайта из нескольких.
vp = engine.volume_probe
good, bad = "162.159.137.232", "104.16.5.5"
stalled = vp.VolumeFacts(vp.VolumeRun(vp.RUN_STALLED, 14_500, 20), vp.VolumeRun(vp.RUN_STALLED, 14_100, 40))
passed = vp.VolumeFacts(vp.VolumeRun(vp.RUN_PASSED, 49_500, 4))
asked: list[str] = []
def volume(host, ip):
asked.append(ip)
return stalled if (host, ip) == ("discord.com", bad) else passed
net = _Net(system=(good, bad), reference=(good, bad))
net.volume_facts = volume
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
service, target = self._discord(result)
self.assertIn(bad, asked)
self.assertEqual((target["state"], target["volume"]), ("ok", "ok"))
self.assertIn(f"по адресу {bad} обрывается", target["unstable"])
self.assertEqual((service["level"], service["headline"]), ("warn", "Discord открывается через раз"))
def test_second_address_of_the_same_network_is_not_asked(self) -> None:
# Соседние адреса одной сети закрыты или открыты все разом: второй раз мерить нечего.
asked: list[str] = []
vp = engine.volume_probe
net = _Net()
net.volume_facts = lambda _host, ip: asked.append(ip) or vp.VolumeFacts(vp.VolumeRun(vp.RUN_PASSED, 49_500, 4))
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
self.assertNotIn(DISCORD_REAL[1], asked)
self.assertEqual(self._discord(result)[1]["unstable"], "")
class SystemStateInReportTests(unittest.TestCase):
@staticmethod
def _item(key, level, text, advice=""):
return engine.system_state.SystemItem(key, f"Заголовок {key}", level, text, advice)
def _run(self, *items, https=None):
lines: list[str] = []
net = _Net(https=https)
net.system_items = tuple(items)
return net.run(engine.run_blockcheck, "main", emit=lines.append), lines
def test_every_item_is_printed_and_returned(self) -> None:
ok = self._item("admin", "ok", "есть")
note = self._item("antivirus", "info", "работает Kaspersky", "Добавьте в исключения")
result, lines = self._run(ok, note)
self.assertIn("━━━━━━━━ Этот компьютер ━━━━━━━━", lines)
self.assertIn(" ✅ Заголовок admin: есть", lines)
self.assertIn(" ℹ️ Заголовок antivirus: работает Kaspersky", lines)
self.assertEqual([item["key"] for item in result["system"]], ["admin", "antivirus"])
self.assertEqual(result["system"][1]["advice"], "Добавьте в исключения")
# Пометки и «всё хорошо» в список проблем не попадают.
self.assertFalse([item for item in result["problems"] if item["text"].startswith("Заголовок")])
def test_failures_and_warnings_become_problems_with_advice(self) -> None:
result, _lines = self._run(
self._item("bfe", "fail", "не работает", "Запустите службу"),
self._item("proxy", "warn", "включён"),
self._item("clock", "unknown", "проверить не удалось"),
)
problems = {item["text"]: item for item in result["problems"] if item["text"].startswith("Заголовок")}
self.assertEqual(set(problems), {"Заголовок bfe: не работает", "Заголовок proxy: включён"})
self.assertEqual(problems["Заголовок bfe: не работает"]["level"], "fail")
self.assertEqual(problems["Заголовок bfe: не работает"]["advice"], ["Запустите службу"])
self.assertEqual(problems["Заголовок proxy: включён"]["advice"], [])
# Неполадка компьютера стоит выше предупреждений о сети.
self.assertEqual(result["problems"][0]["text"], "Заголовок bfe: не работает")
def test_system_problem_is_shown_even_when_nothing_opens(self) -> None:
"""Остановленная служба сама может быть причиной «нет интернета» — молчать о ней нельзя."""
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT))
net.system_items = (self._item("bfe", "fail", "не работает"),)
result = net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
self.assertTrue([item for item in result["problems"] if item["text"] == "Заголовок bfe: не работает"])
def test_no_section_when_nothing_was_collected(self) -> None:
result, lines = self._run()
self.assertNotIn("━━━━━━━━ Состояние системы ━━━━━━━━", lines)
self.assertEqual(result["system"], [])
class ClockSkewTests(unittest.TestCase):
def _skew(self, body: bytes | None, addresses=("142.250.1.1",)):
run = engine._Run(None, workers=4)
self.addCleanup(run.close)
result = ProbeResult(ip="142.250.1.1", kind=KIND_OK, status=204, body=body or b"")
with (
patch.object(net_access, "doh_lookup", return_value=(bool(addresses), tuple(addresses))),
patch.object(net_access, "chrome_get", return_value=result) as get,
patch.object(sections.time, "time", return_value=1_800_000_000.0),
):
return sections.clock_skew(run), get
def test_skew_is_local_time_minus_server_date(self) -> None:
from email.utils import formatdate
body = b"HTTP/1.1 204 No Content\r\nDate: " + formatdate(1_800_000_000 - 7200, usegmt=True).encode() + b"\r\n\r\n"
skew, get = self._skew(body)
self.assertEqual(skew, 7200.0)
self.assertEqual(get.call_args.args[:3], ("www.google.com", "142.250.1.1", "/generate_204"))
def test_no_address_no_date_or_bad_date_is_unknown(self) -> None:
self.assertIsNone(self._skew(b"", addresses=())[0])
self.assertIsNone(self._skew(b"HTTP/1.1 204 No Content\r\nServer: x\r\n\r\n")[0])
self.assertIsNone(self._skew(b"HTTP/1.1 204 No Content\r\nDate: yesterday\r\n\r\n")[0])
# Строка «Date:» в теле ответа — не заголовок.
self.assertIsNone(self._skew(b"HTTP/1.1 200 OK\r\nServer: x\r\n\r\nDate: Mon, 01 Jan 2024 00:00:00 GMT")[0])
class FullCheckTests(unittest.TestCase):
"""«Глубокая диагностика»: всё из «Всех сайтов» плюс DNS-серверы, место фильтра и пробы, которых браузер не делает."""
DNS_OK = {
"level": "warn",
"findings": [
{"level": "fail", "text": "Обычные DNS-запросы перехватываются по дороге."},
{"level": "info", "text": "Не отвечают совсем: Xbox DNS (old)."},
{"level": "ok", "text": "Для защищённого DNS сейчас лучше всего подходит Cloudflare (1.1.1.1)."},
],
"text": "Сервер Адрес\nCloudflare 1.1.1.1",
}
def _run(self, scope="full", *, dns=None, quic_blocked=(), trace=None, filter_facts=None, https=None):
from diagnostics import path_trace
lines: list[str] = []
calls: dict = {"dns": 0, "trace": [], "locate": []}
def check_dns(*, should_stop=None):
calls["dns"] += 1
return dns if dns is not None else self.DNS_OK
def fake_trace(ip, **_kwargs):
calls["trace"].append(ip)
return trace or path_trace.RouteTrace(target=ip, supported=False)
def fake_locate(ip, name, *, max_ttl, cancel, **_kwargs):
calls["locate"].append((ip, name, max_ttl))
return filter_facts or path_trace.FilterFacts(True, True, 6, 6)
net = _Net(https=https)
net.quic_facts = lambda host, ip: (
engine.quic_probe.QuicFacts(host=host, neutral_ms=40.0)
if host in quic_blocked
else engine.quic_probe.QuicFacts(host=host, real_ms=10.0)
)
from contextlib import ExitStack
with ExitStack() as stack:
for item in net.patches():
stack.enter_context(item)
stack.enter_context(patch.object(path_trace, "trace_route", fake_trace))
stack.enter_context(patch.object(path_trace, "locate_filter", fake_locate))
result = engine.run_blockcheck(scope, emit=lines.append, check_dns_servers=check_dns)
return result, lines, calls
def test_full_scope_checks_all_sites_and_dns_servers(self) -> None:
result, lines, calls = self._run()
self.assertEqual(result["scope"], "full")
self.assertIn("🔍 BlockCheck: Глубокая диагностика", lines[0])
self.assertIn("telegram", {item["key"] for item in result["services"]})
self.assertEqual(calls["dns"], 1)
self.assertEqual(result["dns_servers"]["level"], "warn")
self.assertIn("━━━━━━━━ DNS-серверы ━━━━━━━━", lines)
self.assertIn("❌ Обычные DNS-запросы перехватываются по дороге.", lines)
self.assertIn(" Cloudflare 1.1.1.1", lines)
def test_dns_findings_that_matter_become_problems_with_dns_button(self) -> None:
result, _lines, _calls = self._run()
dns_problems = [item for item in result["problems"] if "DNS" in item["text"] and item["action"] == "dns"]
self.assertEqual([item["text"] for item in dns_problems], ["Обычные DNS-запросы перехватываются по дороге."])
self.assertEqual(dns_problems[0]["level"], "fail")
# Пометки и «лучший сервер» в список проблем не идут.
self.assertFalse([item for item in result["problems"] if "Xbox DNS" in item["text"] or "лучше всего" in item["text"]])
def test_dns_finding_parts_reach_the_report_and_the_problem(self) -> None:
spoofed = {
"level": "fail",
"text": "Обычные ответы подменяются у серверов: Cloudflare (1.1.1.1) и ещё 1.",
"title": "Обычные ответы подменяются у серверов",
"servers": [["Cloudflare", "1.1.1.1"], ["Cloudflare", "1.0.0.1"]],
"note": "",
}
plain = {"level": "warn", "text": "DNS без частей: старая проверка."}
result, _lines, _calls = self._run(dns={"level": "fail", "findings": [spoofed, plain], "text": ""})
self.assertEqual(result["dns_servers"]["findings"], [spoofed, plain])
with_parts, without = [item for item in result["problems"] if item["action"] == "dns"]
# Части лежат отдельным ключом: «title» самой проблемы — это название сайта.
self.assertEqual(with_parts["title"], "")
self.assertEqual(
with_parts["parts"],
{"title": spoofed["title"], "servers": spoofed["servers"], "note": ""},
)
self.assertNotIn("parts", without)
def test_filter_place_is_searched_for_site_with_quic_blocked_by_name(self) -> None:
from diagnostics import path_trace
from utils.windows_icmp import HOP_ROUTER
trace = path_trace.RouteTrace(
DISCORD_REAL[0], tuple(path_trace.Hop(ttl, HOP_ROUTER, f"10.0.0.{ttl}", 1.0) for ttl in range(1, 8)), reached=True
)
result, lines, calls = self._run(quic_blocked={"discord.com"}, trace=trace)
self.assertEqual(calls["locate"], [(DISCORD_REAL[0], "discord.com", sections.FILTER_MAX_TTL)])
self.assertEqual(calls["trace"], [DISCORD_REAL[0]])
place = result["filter"]
self.assertEqual((place["host"], place["found"], place["hop"], place["state"]), ("discord.com", True, 6, "found"))
self.assertIn("Фильтр стоит между узлами 5 и 6 от вас — в сети вашего провайдера", place["text"])
# Сайт один, и во время поиска работал Zapret: уверенность низкая, и об этом сказано.
self.assertEqual(place["confidence"], "low")
self.assertIn("Zapret", " ".join(place["reasons"]))
self.assertIn("срок жизни (TTL)", place["ttl_advice"])
self.assertEqual([hop["owner"] for hop in place["hops"][:2]], ["ваш роутер", "внутренняя сеть провайдера"])
self.assertEqual(len(place["hops"]), 7)
text = "\n".join(lines)
self.assertIn("━━━━━━━━ Место фильтра ━━━━━━━━", text)
self.assertIn("── здесь стоит фильтр ──", text)
def test_without_blocks_by_name_the_road_is_still_shown(self) -> None:
from diagnostics import path_trace
from utils.windows_icmp import HOP_ROUTER
trace = path_trace.RouteTrace("1.1.1.1", tuple(path_trace.Hop(ttl, HOP_ROUTER, f"10.0.0.{ttl}", 1.0) for ttl in range(1, 4)))
result, lines, calls = self._run(trace=trace)
self.assertEqual((calls["locate"], len(calls["trace"])), ([], 1))
place = result["filter"]
self.assertEqual((place["state"], place["found"], len(place["hops"])), ("none", False, 3))
self.assertTrue(place["host"])
self.assertIn("искать место фильтра не по чему", place["text"])
def test_filter_is_searched_on_several_sites_but_one_address_only_once(self) -> None:
_result, _lines, calls = self._run(quic_blocked={"discord.com", "www.youtube.com", "telegram.org"})
addresses = [ip for ip, _name, _ttl in calls["locate"]]
self.assertEqual(len(addresses), len(set(addresses)))
self.assertLessEqual(len(addresses), engine.sections.filter_place.MAX_SITES)
def test_other_scopes_do_not_run_the_extra_checks(self) -> None:
for scope in ("all", "main"):
with self.subTest(scope=scope):
result, lines, calls = self._run(scope, quic_blocked={"discord.com"})
self.assertEqual((calls["dns"], calls["locate"]), (0, []))
self.assertIsNone(result["dns_servers"])
self.assertIsNone(result["filter"])
def test_unknown_scope_falls_back_to_main(self) -> None:
result, _lines, _calls = self._run("что-то")
self.assertEqual(result["scope"], "main")
def test_broken_dns_check_does_not_break_the_rest(self) -> None:
lines: list[str] = []
def broken(*, should_stop=None):
raise OSError("нет сети")
net = _Net()
result = net.run(engine.run_blockcheck, "full", emit=lines.append, check_dns_servers=broken)
self.assertIsNone(result["dns_servers"])
self.assertIn("❔ DNS-серверы: проверка не выполнилась (нет сети)", lines)
self.assertIn("Discord", result["working"])
def test_full_check_without_dns_function_still_runs(self) -> None:
net = _Net()
result = net.run(engine.run_blockcheck, "full", emit=lambda _line: None)
self.assertEqual(result["scope"], "full")
self.assertIsNone(result["dns_servers"])
class EngineScenarioTests(unittest.TestCase):
"""Движок целиком, сеть подменена фейками."""
def test_unblocking_dns_is_not_reported_as_spoofing(self) -> None:
lines: list[str] = []
net = _Net(system=UNBLOCK_PROXY)
result = net.run(engine.run_dns_check, emit=lines.append)
text = "\n".join(lines)
self.assertFalse(result["summary"]["dns_poisoning_detected"])
self.assertNotIn("Обнаружена DNS подмена", text)
self.assertIn("Comss DNS", text)
self.assertIn("сам меняет адреса", text)
def test_dns_that_answers_nxdomain_only_sometimes_is_still_spoofing(self) -> None:
"""У пользователя вкладка DNS то находила подмену, то нет: провайдер отвечал «сайта нет» не каждый раз."""
answers = iter([
DnsAnswer(ips=(), status=DNS_STATUS_NAME_ERROR),
DnsAnswer(ips=DISCORD_REAL),
DnsAnswer(ips=DISCORD_REAL),
] * 10)
lines: list[str] = []
net = _Net()
with patch.object(net_access, "query_ipv4", side_effect=lambda *_a, **_k: next(answers)):
patches = [item for item in net.patches() if getattr(item, "attribute", "") != "query_ipv4"]
from contextlib import ExitStack
with ExitStack() as stack:
for item in patches:
stack.enter_context(item)
result = engine.run_dns_check(emit=lines.append)
self.assertTrue(result["summary"]["dns_poisoning_detected"])
self.assertIn("то правильно, то «такого сайта нет»", "\n".join(lines))
def test_one_nxdomain_among_answers_without_any_address_is_still_spoofing(self) -> None:
# Серии [молчит, «нет сайта», молчит] и [пусто, «нет сайта», пусто]: адреса не было
# ни разу, а первый ответ не NXDOMAIN. Правило «хоть один NXDOMAIN» смягчать нельзя.
from diagnostics.verdict import judge_dns
# 1460 — истекло время ожидания, 9002 — отказ сервера (SERVFAIL).
for first in (1460, 9002):
with self.subTest(first_status=first):
judgement = judge_dns(
system_ips=(),
system_status=first,
reference_ips=DISCORD_REAL,
hosts_ips=(),
nxdomain_count=1,
attempts=3,
)
self.assertEqual(judgement.state, DnsState.SPOOFED)
self.assertIn("1 из 3", judgement.reason)
def test_silence_without_any_nxdomain_stays_unknown(self) -> None:
from diagnostics.verdict import judge_dns
judgement = judge_dns(system_ips=(), system_status=1460, reference_ips=DISCORD_REAL, hosts_ips=(), attempts=3)
self.assertEqual(judgement.state, DnsState.UNKNOWN)
def test_dns_tab_skips_https_when_address_matches_reference(self) -> None:
net = _Net()
net.run(engine.run_dns_check, emit=lambda _line: None)
self.assertEqual(net.calls, [])
def test_foreign_certificate_is_reported_as_spoofing(self) -> None:
# Чужой сертификат — только по адресу из DNS; настоящий адрес сайта отдаёт подлинный.
lines: list[str] = []
net = _Net(
system=("5.6.7.8",),
https=lambda host, ip: (
ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан другому сайту") if ip == "5.6.7.8" else _ok(ip)
),
)
result = net.run(engine.run_dns_check, emit=lines.append)
self.assertTrue(result["summary"]["dns_poisoning_detected"])
self.assertIn("Обнаружена DNS подмена", "\n".join(lines))
# Контроль спрашивал настоящий адрес сайта.
self.assertIn(DISCORD_REAL[0], [ip for _host, ip in net.calls])
def test_foreign_certificate_everywhere_is_not_dns_spoofing(self) -> None:
# Антивирус или прокси с проверкой HTTPS: чужой сертификат на любом адресе — DNS ни при чём.
lines: list[str] = []
net = _Net(
system=("5.6.7.8",),
https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан другому сайту"),
)
result = net.run(engine.run_dns_check, emit=lines.append)
self.assertFalse(result["summary"]["dns_poisoning_detected"])
self.assertIn("DNS тут ни при чём", "\n".join(lines))
def test_foreign_certificate_without_reference_proves_nothing(self) -> None:
from diagnostics.verdict import judge_dns
judgement = judge_dns(
system_ips=("5.6.7.8",), system_status=0, reference_ips=(), hosts_ips=(), check_kind=KIND_CERT
)
self.assertEqual(judgement.state, DnsState.UNKNOWN)
self.assertIn("эталон недоступен", judgement.reason)
def test_nxdomain_site_that_opens_by_real_address_is_reported_as_working(self) -> None:
"""Случай из жалобы: DNS говорит «домена нет», браузер с DoH открывает сайт.
Итог не красный и не «заблокирован»: сайт доступен. Но и не голое «открывается» —
браузер без защищённого DNS его не найдёт (жалоба 2026-10-13 «проверка проходит, а сайт не открывается»).
"""
lines: list[str] = []
net = _Net(system=(), status=DNS_STATUS_NAME_ERROR, reference=("142.251.157.4",))
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
text = "\n".join(lines)
youtube = next(item for item in result["services"] if item["key"] == "youtube")
self.assertIn("YouTube: сам сайт доступен, но DNS компьютера отвечает, что такого сайта нет", youtube["headline"])
self.assertIn("браузер без защищённого DNS", youtube["headline"])
self.assertEqual(youtube["level"], "warn")
self.assertIn("www.youtube.com", youtube["dns_note"])
self.assertTrue(result["spoofed_hosts"])
self.assertIn(("www.youtube.com", "142.251.157.4"), net.calls)
self.assertNotIn("❌ YouTube", text)
def test_report_grows_during_the_check_and_the_last_partial_is_the_final_one(self) -> None:
# Экран рисует разделы по мере готовности. Сервисы приходят по одному, а не разом в конце;
# последний опубликованный отчёт — это и есть итоговый.
seen: list[dict] = []
final = _Net().run(engine.run_blockcheck, "main", emit=lambda _line: None, partial=seen.append)
counts = [len(item["services"] or ()) for item in seen]
self.assertIn(1, counts)
self.assertEqual(counts, sorted(counts))
self.assertTrue(all(item["partial"] for item in seen[:-1]))
self.assertEqual(seen[-1], final)
self.assertFalse(final["partial"])
# У каждого адреса записано, сколько секунд на него ушло, — и это видно в подробностях сайта.
target = final["services"][0]["targets"][0]
self.assertGreaterEqual(target["seconds"], 0.0)
from blockcheck.ui.result_cards_model import build_cards
timed = {**final, "services": [{**final["services"][0], "targets": [{**target, "seconds": 12.4}]}]}
rows = [line for section in build_cards(timed)[0].sections for line in section.lines]
self.assertIn(("Проверка заняла", "12 с"), [(line.name, line.text) for line in rows])
# Набор разделов один и тот же с первого отчёта: экран не гадает, каких ключей ещё нет.
self.assertLessEqual(set(engine._REPORT_SECTIONS), set(seen[0]))
def test_vpn_caveat_is_said_once_above_not_in_every_row(self) -> None:
# О запущенных VPN итог говорит одной строкой над проблемами (bypass_warning/bypass_note),
# а не повтором в каждой строке: в отчёте есть список программ, в советах сайтов — нет.
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET))
net.bypass_tools = ("sing-box", "WireGuard")
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertEqual(result["other_bypass_tools"], ["sing-box", "WireGuard"])
self.assertFalse([line for item in result["problems"] for line in item["advice"] if "sing-box" in line])
def test_dpi_reset_retries_once_on_another_address_and_reports_strategy(self) -> None:
lines: list[str] = []
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET))
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
text = "\n".join(lines)
discord_calls = [ip for host, ip in net.calls if host == "discord.com"]
# Оба адреса пробуются в общем залпе; при повторной проверке — один лёгкий запрос
# к тому адресу, на котором сайт не открылся: полный повтор нужен, только если он ожил.
self.assertEqual(len(discord_calls), len(DISCORD_REAL) + 1)
self.assertEqual(set(discord_calls), set(DISCORD_REAL))
discord_target = result["services"][0]["targets"][0]
self.assertEqual(discord_target["rechecked"], "same")
self.assertIn("повторная проверка поодиночке дала то же", discord_target["text"])
self.assertFalse(result["spoofed_hosts"])
self.assertEqual(result["services"][0]["level"], "fail")
self.assertIn("Подбор стратегии", text)
# Заголовок раздела адреса: «имя — для чего»; порядок разделов — порядок проверки.
hosts_order = [line.split(" ")[0] for line in lines if re.match(r"[a-z0-9.-]+\.[a-z]+ — ", line)]
self.assertEqual(hosts_order[:3], ["discord.com", "gateway.discord.gg", "cdn.discordapp.com"])
def test_site_that_opens_on_the_calm_second_check_is_not_reported_as_blocked(self) -> None:
"""Первый сбой дала нагрузка самой проверки: при повторе поодиночке сайт открывается."""
seen: dict[str, int] = {}
def _https(host, ip):
seen[host] = seen.get(host, 0) + 1
# discord.com не соединяется в залпе (оба адреса и повтор после паузы), потом открывается.
if host == "discord.com" and seen[host] <= 3:
return ProbeResult(ip=ip, kind=KIND_CONNECT, connect_fail="timeout")
return _ok(ip)
net = _Net(https=_https)
with patch.object(reach, "RETRY_PAUSE_S", 0.0):
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertEqual(discord["level"], "ok")
self.assertEqual(discord["targets"][0]["rechecked"], "opened")
self.assertIn("со второй проверки, поодиночке", discord["targets"][0]["text"])
self.assertFalse(any(item["target"] == "discord.com" for item in result["problems"]))
def test_dns_tab_does_not_recheck(self) -> None:
"""Вкладка «DNS подмена» проверяет DNS, а не открываемость: повторный проход ей не нужен."""
calls = []
def _https(host, ip):
calls.append(host)
return ProbeResult(ip=ip, kind=KIND_RESET)
net = _Net(system=("5.6.7.8",), https=_https)
net.run(engine.run_dns_check, emit=lambda _line: None)
self.assertEqual(len(calls), len(set(calls)))
def test_site_blocked_over_ipv4_but_open_over_ipv6_is_reported_as_working(self) -> None:
"""Браузер сам уходит на IPv6, поэтому и проверка обязана его попробовать."""
v6 = "2606:4700::6810:1"
def _https(host, ip):
return _ok(ip) if ip == v6 else ProbeResult(ip=ip, kind=KIND_RESET)
net = _Net(https=_https, reference_v6=(v6,))
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertEqual(discord["level"], "ok")
self.assertIn("IPv6", discord["targets"][0]["short"])
self.assertIn(("discord.com", v6), net.calls)
def test_site_down_from_abroad_too_is_not_called_a_block(self) -> None:
"""Сайт без дороги, который не открылся и с сервера за рубежом, — не блокировка."""
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "discord.com" else _ok(ip))
net.outside = {"discord.com": "down"}
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None, ask_outside=True)
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertEqual((discord["kind"], discord["level"]), ("down", "warn"))
self.assertIn("Zapret здесь не поможет", discord["advice"][0])
self.assertIn("discord.com", net.outside_asked)
# Кнопки «Подобрать стратегию» нет: стратегия лежащий сайт не поднимет.
problem = next(item for item in result["problems"] if "Discord" in item["text"])
self.assertNotIn(problem["action"], ("strategy", "start_zapret"))
def test_server_abroad_is_not_asked_without_the_checkbox(self) -> None:
# Соединение с сервером проекта выдаёт программу провайдеру: без галочки его нет.
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "discord.com" else _ok(ip))
net.outside = {"discord.com": "down"}
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertEqual(net.outside_asked, [])
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertNotEqual(discord["kind"], "down")
def test_own_domains_and_control_sites_are_not_sent_abroad(self) -> None:
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET))
net.run(engine.run_blockcheck, "all", user_domains=("my-own.example",), emit=lambda _line: None, ask_outside=True)
self.assertNotIn("my-own.example", net.outside_asked)
self.assertFalse({"google.com", "www.google.com", "cloudflare.com", "ya.ru", "vk.com"} & set(net.outside_asked))
def test_ipv6_is_not_tried_when_ipv4_opens(self) -> None:
net = _Net(reference_v6=("2606:4700::6810:1",))
net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertFalse(any(":" in ip for _host, ip in net.calls))
# Адреса IPv6 нужны только запасной попытке — у открывшегося сайта их не спрашивают.
self.assertNotIn(TYPE_AAAA, net.doh_types)
def test_system_address_that_fails_is_rechecked_by_reference_address(self) -> None:
def _https(host, ip):
if ip == "5.6.7.8":
return ProbeResult(ip=ip, kind=KIND_RESET)
return _ok(ip)
net = _Net(system=("5.6.7.8",), https=_https)
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertEqual(result["services"][0]["level"], "ok")
self.assertIn(("discord.com", DISCORD_REAL[0]), net.calls)
def test_run_deadline_still_prints_summary_and_keeps_spoofing(self) -> None:
# Запас по времени большой: при загруженной машине ответы DNS не должны
# опоздать к сроку — иначе тест падает через раз.
def _slow(host, ip):
time.sleep(2.0)
return ProbeResult(ip=ip, kind="cancelled")
lines: list[str] = []
net = _Net(system=("195.82.146.214",), https=_slow)
with patch.object(engine, "RUN_DEADLINE", 1.0):
result = net.run(engine.run_dns_check, emit=lines.append)
text = "\n".join(lines)
self.assertNotIn("stopped", result)
self.assertTrue(result["summary"]["dns_poisoning_detected"])
self.assertIn("Итог", text)
self.assertIn("не уложилась", text)
def test_stop_returns_quickly_and_marks_stopped(self) -> None:
release = threading.Event()
def _slow(host, ip):
release.wait(5)
return _ok(ip)
net = _Net(system=("5.6.7.8",), https=_slow)
started = time.monotonic()
result = net.run(engine.run_dns_check, emit=lambda _line: None, should_stop=lambda: True)
release.set()
self.assertTrue(result.get("stopped"))
self.assertLess(time.monotonic() - started, 2.0)
class EngineMixedAnswerTests(unittest.TestCase):
"""Найденные при разборе ошибки: смешанный ответ DNS, прерывание по времени, hosts."""
def _mixed_net(self, https):
stub = "95.1.1.1"
# У каждого сайта своя очередь ответов: с общей очередью потоки делили бы
# ответы как придётся, и «чужой» адрес доставался бы не тому сайту.
lock = threading.Lock()
queues: dict[str, object] = {}
def _system(host):
with lock:
answers = queues.setdefault(
host,
iter([DnsAnswer(ips=(stub,)), DnsAnswer(ips=DISCORD_REAL), DnsAnswer(ips=DISCORD_REAL)]),
)
return next(answers)
return _Net(system=_system, https=https), stub
def test_dns_that_sometimes_gives_foreign_server_is_spoofing_but_site_opens(self) -> None:
def _https(host, ip):
if ip == "95.1.1.1":
return ProbeResult(ip=ip, kind=KIND_CERT, cert_problem="сертификат выдан другому сайту")
return _ok(ip)
net, _stub = self._mixed_net(_https)
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["services"] if item["key"] == "discord")
main = next(item for item in discord["targets"] if item["main"])
# Сайт открывается по настоящему адресу, а DNS помечен как подмена.
self.assertTrue(main["ok"])
self.assertEqual(main["dns_state"], "spoofed")
self.assertIn("чужого сервера", main["dns_reason"])
def test_cdn_with_extra_genuine_address_is_not_spoofing(self) -> None:
net, _stub = self._mixed_net(lambda host, ip: _ok(ip))
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertTrue(all(item["dns_state"] == "ok" for item in discord["targets"]))
def test_deadline_before_first_request_is_not_no_address(self) -> None:
probe = engine._Probe(target=engine.Target("discord.com", "сайт"), service="discord", host="discord.com")
probe.dns = DnsAnswer(ips=DISCORD_REAL)
probe.reference_ips = DISCORD_REAL
run = engine._Run(None, workers=1, deadline=0)
self.addCleanup(run.close)
reach.check_reach(run, probe, read_limit=0)
self.assertEqual(judge_reach(probe.reach), ReachState.UNKNOWN)
def test_stale_hosts_entry_gets_hosts_advice_not_antivirus(self) -> None:
verdict = summarize_service(
"Instagram",
[TargetOutcome("www.instagram.com", "сайт", ReachState.CERT, DnsState.LOCAL, main=True)],
zapret_running=True,
)
self.assertIn("hosts", verdict.headline)
self.assertTrue(any("hosts" in item for item in verdict.advice))
self.assertFalse(any("антивирус" in item for item in verdict.advice))
def test_stale_hosts_advice_is_kept_when_another_address_is_blocked_too(self) -> None:
verdict = summarize_service(
"Discord",
[
TargetOutcome("discord.com", "сайт", ReachState.OK, DnsState.LOCAL, main=True, hosts_stale=True),
TargetOutcome("cdn.discordapp.com", "картинки", ReachState.DPI, DnsState.OK),
],
zapret_running=True,
)
self.assertIn("не работают картинки", verdict.headline)
self.assertIn(ADVICE_HOSTS_STALE, verdict.advice)
def test_user_domains_only_in_all_sites_scope(self) -> None:
self.assertNotIn("user:example.org", engine.build_services("main", ["example.org"]))
self.assertIn("user:example.org", engine.build_services("all", ["example.org"]))
class BlockcheckScopeTests(unittest.TestCase):
"""Режимы BlockCheck, голосовые серверы, обрыв 16 КБ и сводка проблем."""
def test_scopes_and_user_domains(self) -> None:
main = engine.build_services("main")
everything = engine.build_services("all", ["Example.org", "discord.com"])
self.assertEqual(list(main), ["discord", "youtube"])
self.assertIn("telegram", everything)
self.assertTrue(everything["google"].control)
self.assertIn("user:example.org", everything)
# Домен, который и так проверяется, не дублируется.
self.assertNotIn("user:discord.com", everything)
def test_unparsable_stun_answer_means_udp_works(self) -> None:
from types import SimpleNamespace
from diagnostics.voice_check import _describe, summarize_voice, VoiceServer
answered, _text, decided = _describe(SimpleNamespace(status=SimpleNamespace(value="fail"), error_code="PARSE_ERR", detail=""))
self.assertTrue(answered and decided)
# Адрес сервера узнать не удалось — это не «UDP закрыт»: такие серверы в вывод не идут.
_answered, _text, decided = _describe(SimpleNamespace(status=SimpleNamespace(value="fail"), error_code="DNS_ERR", detail=""))
self.assertFalse(decided)
unresolved = VoiceServer("A", "a", False, "не удалось узнать адрес", decided=False)
self.assertEqual(summarize_voice((unresolved, unresolved)).level, Level.UNKNOWN)
self.assertEqual(summarize_voice((unresolved, VoiceServer("B", "b", True, "да"))).level, Level.OK)
report = summarize_voice((VoiceServer("A", "a", False, "нет"), VoiceServer("B", "b", False, "нет")))
self.assertEqual(report.level, Level.FAIL)
# На win10 молчали только серверы Telegram — это про звонки Telegram, не «UDP вообще».
report = summarize_voice((
VoiceServer("Google STUN", "stun.l.google.com", True, "да"),
VoiceServer("Telegram STUN", "stun.telegram.org", False, "нет", group="telegram"),
))
self.assertEqual(report.level, Level.WARN)
self.assertIn("Telegram", report.headline)
# Telegram узнаётся по группе из списка серверов, а не по слову в имени.
report = summarize_voice((
VoiceServer("Google STUN", "stun.l.google.com", True, "да"),
VoiceServer("Мой telegram-подобный", "stun.example", False, "нет"),
))
self.assertEqual(report.level, Level.OK)
# «Нет дороги до адреса» — не «UDP закрыт»: такой сервер в вывод не идёт.
_answered, _text, decided = _describe(SimpleNamespace(status=SimpleNamespace(value="error"), error_code="UNREACHABLE", detail=""))
self.assertFalse(decided)
report = summarize_voice((
VoiceServer("Google STUN", "stun.l.google.com", True, "да"),
VoiceServer("CF STUN", "stun.cloudflare.com", False, "нет"),
))
self.assertEqual(report.level, Level.OK)
def test_stun_classifies_windows_codes(self) -> None:
import socket as _socket
from blockcheck import stun_tester
def unreachable(*_a, **_k):
error = OSError(None, "сообщение на языке системы")
error.winerror = 10065
raise error
class _Sock:
def __init__(self, *_a):
pass
def settimeout(self, _t):
pass
sendto = unreachable
def close(self):
pass
address = (_socket.AF_INET, _socket.SOCK_DGRAM, 0, ("192.0.2.1", 3478))
with patch.object(stun_tester, "_resolve_udp_addresses", return_value=[address]), patch.object(
stun_tester.socket, "socket", _Sock
):
result = stun_tester.test_stun("stun.example", 3478, timeout=1, retries=1)
self.assertEqual(result.error_code, "UNREACHABLE")
def test_every_stun_server_names_its_group(self) -> None:
from blockcheck.data_lists import STUN_TARGETS
self.assertTrue(all(item.get("group") in ("general", "telegram") for item in STUN_TARGETS))
self.assertTrue(any(item["group"] == "telegram" for item in STUN_TARGETS))
def test_freeze_classification(self) -> None:
from diagnostics.freeze_check import FreezeState, classify_download
frozen = classify_download(_ok(body_cut=True, body_size=16_500))
self.assertEqual(frozen[0], FreezeState.FREEZE)
self.assertEqual(classify_download(_ok(body_size=65_536))[0], FreezeState.OK)
# Сброс сразу, без данных — не «доступ есть» и не «обрыв», а «не удалось проверить».
self.assertEqual(classify_download(ProbeResult(ip="1.1.1.1", kind=KIND_RESET))[0], FreezeState.UNKNOWN)
self.assertEqual(classify_download(_ok(body_size=2_000))[0], FreezeState.UNKNOWN)
def test_freeze_takes_spare_address_of_same_provider(self) -> None:
from diagnostics import freeze_check
from diagnostics.freeze_check import FreezeState, check_freeze
targets = (
{"provider": "Hetzner", "url": "https://dead.example/a.css"},
{"provider": "Hetzner", "url": "https://alive.example/b.bin"},
{"provider": "Hetzner", "url": "https://third.example/c.bin"},
{"provider": "OVH", "url": "https://ovh.example/1Mb.dat"},
)
calls: list[str] = []
def _download(host, _path):
calls.append(host)
return None if host == "dead.example" else _ok(body_size=65_000)
with patch("blockcheck.data_lists.TCP_16_20_TARGETS", targets):
servers = check_freeze(lambda fn, *a: _Done(fn(*a)), lambda f: f.result(), _download)
self.assertEqual([item.state for item in servers], [FreezeState.OK, FreezeState.OK])
self.assertIn("alive.example", servers[0].name)
# Ответ получен со второго адреса — третий не трогаем.
self.assertNotIn("third.example", calls)
# Вышел запас времени — запасные адреса не пробуются.
calls.clear()
with patch("blockcheck.data_lists.TCP_16_20_TARGETS", targets):
servers = check_freeze(lambda fn, *a: _Done(fn(*a)), lambda f: f.result(), _download, budget=0)
self.assertEqual(servers[0].state, FreezeState.UNKNOWN)
self.assertNotIn("alive.example", calls)
self.assertEqual(freeze_check.CANDIDATES_PER_PROVIDER, 3)
def test_freeze_summary_says_how_many_servers_were_checked(self) -> None:
from diagnostics.freeze_check import FreezeServer, FreezeState, summarize_freeze
ok = FreezeServer("A", FreezeState.OK, "32 КБ")
unknown = FreezeServer("B", FreezeState.UNKNOWN, "сброс")
full = summarize_freeze((ok, ok), zapret_running=True)
self.assertEqual((full.level, full.headline), (Level.OK, "Обрыва загрузки на 16–20 КБ нет"))
half = summarize_freeze((ok, ok, ok, unknown, unknown, unknown), zapret_running=True)
self.assertEqual(half.level, Level.OK)
self.assertIn("3 из 6", half.headline)
mostly_unknown = summarize_freeze((ok, unknown, unknown), zapret_running=True)
self.assertEqual(mostly_unknown.level, Level.UNKNOWN)
self.assertIn("1 из 3", mostly_unknown.headline)
def _run_all(self, https, *, subnets=None, net_out: list | None = None):
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.voice_check import VoiceServer
net = _Net(
https=https,
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
net.subnets = subnets
if net_out is not None:
net_out.append(net)
return net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
def _run_with_telegram(self, connected, *, scope="all"):
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.telegram_check import DATA_CENTERS, DcResult
from diagnostics.voice_check import VoiceServer
net = _Net(
https=lambda host, ip: _ok(ip),
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
net.telegram = tuple(
DcResult(center, ok, 30.0 if ok else None, attempts=1 if ok else 2)
for center, ok in zip(DATA_CENTERS, connected)
)
net.network = {"external_ip": "203.0.113.7", "provider": "ROSTELECOM-AS · AS12389", "lines": []}
return net.run(engine.run_blockcheck, scope, emit=lambda _line: None)
def test_silent_telegram_data_centres_become_a_problem_only_when_all_are_silent(self) -> None:
dead = self._run_with_telegram([False] * 5)
texts = [item["text"] for item in dead["problems"]]
self.assertTrue(any("Дата-центры Telegram не принимают соединения" in text for text in texts))
self.assertEqual(dead["telegram"]["level"], "fail")
self.assertEqual([item["state"] for item in dead["telegram"]["items"]], ["fail"] * 5)
self.assertEqual(dead["telegram"]["items"][0]["text"], "не соединился, попыток: 2")
partial = self._run_with_telegram([True, True, True, True, False])
self.assertEqual(partial["telegram"]["level"], "warn")
self.assertFalse(any("Telegram" in item["text"] and "дата-центр" in item["text"].lower() for item in partial["problems"]))
def test_network_block_reaches_the_report_and_quick_check_skips_telegram(self) -> None:
full = self._run_with_telegram([True] * 5)
self.assertEqual(full["network"]["provider"], "ROSTELECOM-AS · AS12389")
self.assertEqual(full["telegram"]["items"][0]["text"], "соединение за 30 мс")
quick = self._run_with_telegram([False] * 5, scope="main")
self.assertIsNone(quick["telegram"])
self.assertEqual(quick["network"]["external_ip"], "203.0.113.7")
def test_cut_on_foreign_controls_is_not_whitelist_mode(self) -> None:
"""Google и Cloudflare обрываются после 16 КБ — это лечит Zapret, а не «закрыты все зарубежные адреса»."""
domestic = ("ya.ru", "vk.com")
def https(host, ip):
if host in domestic:
return _ok(ip)
return ProbeResult(ip=ip, kind="ok", status=200, body_size=16_500, body_cut=True)
result = self._run_all(https)
kinds = [item["kind"] for item in result["problems"]]
self.assertNotIn("network", kinds)
self.assertFalse(any("белых списков" in item["text"] for item in result["problems"]))
# И остальные находки не спрятаны «общей причиной».
self.assertTrue(any(item["kind"] == "cut16" for item in result["problems"]))
def test_answering_reference_server_refutes_no_internet_and_whitelist(self) -> None:
"""Эталонный DNS-сервер — зарубежный адрес: раз он ответил, интернет есть и зарубежные адреса открыты."""
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.voice_check import VoiceServer
net = _Net(
https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT, connect_fail="timeout"),
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
reference = [{"label": "Cloudflare", "address": "1.1.1.1", "ok": True, "answered": 5, "failed": 0, "reason": ""}]
with patch.object(engine._Run, "reference_report", return_value=reference):
result = net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
self.assertFalse(any(item["kind"] == "network" for item in result["problems"]))
def _run_with_quic(self, code_for):
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.voice_check import VoiceServer
net = _Net(
https=lambda host, ip: _ok(ip),
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
quic = engine.quic_probe
def facts(host, _ip):
code = code_for(host)
if code == quic.QUIC_OK:
return quic.QuicFacts(host=host, real_ms=10.0)
return quic.QuicFacts(host=host)
net.quic_facts = facts
return net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
def test_silent_quic_on_sites_that_surely_support_it_means_udp_443_is_closed(self) -> None:
quic = engine.quic_probe
closed = self._run_with_quic(lambda _host: quic.QUIC_SILENT)
[problem] = [item for item in closed["problems"] if item["kind"] == "quic"]
self.assertIn("UDP 443 закрыт целиком", problem["text"])
self.assertIn("www.google.com", problem["text"])
def test_one_answering_site_means_udp_443_is_not_closed(self) -> None:
quic = engine.quic_probe
result = self._run_with_quic(lambda host: quic.QUIC_OK if host == "www.google.com" else quic.QUIC_SILENT)
self.assertFalse(any("закрыт целиком" in item["text"] for item in result["problems"]))
def test_speed_is_measured_only_in_the_full_check(self) -> None:
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.voice_check import VoiceServer
def run(scope):
net = _Net(
https=lambda host, ip: _ok(ip),
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
net.speed = {"level": "ok", "headline": "Заметной разницы в скорости нет", "items": []}
return net.run(engine.run_blockcheck, scope, emit=lambda _line: None)
self.assertEqual(run("full")["speed"]["level"], "ok")
self.assertIsNone(run("all")["speed"])
def test_controls_down_means_no_internet_first(self) -> None:
result = self._run_all(lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT))
self.assertIn("контрольные сайты", result["problems"][0]["text"])
def test_only_domestic_sites_open_means_whitelist_mode(self) -> None:
domestic = {"ya.ru", "vk.com"}
def _https(host, ip):
return _ok(ip) if host in domestic else ProbeResult(ip=ip, kind=KIND_CONNECT)
result = self._run_all(_https)
first = result["problems"][0]
self.assertEqual(first["level"], "fail")
self.assertIn("белых списков", first["text"])
self.assertIn("Яндекс, ВКонтакте", first["text"])
self.assertIn("Google, Cloudflare", first["text"])
self.assertIn("Zapret не помогает", first["advice"][0])
self.assertNotIn("контрольные сайты", first["text"].split("зарубежные")[0])
# Причина одна и названа: кнопок «Подобрать стратегию» у каждого сайта нет.
self.assertFalse([item for item in result["problems"] if item["action"] in ("strategy", "start_zapret")])
yandex = next(item for item in result["services"] if item["key"] == "yandex")
self.assertTrue(yandex["control"] and yandex["domestic"])
def test_subnets_confirm_or_refute_the_whitelist(self) -> None:
"""Признак по контрольным сайтам проверяется по десяткам сетей: они его подтверждают или снимают."""
from diagnostics import subnet_whitelist as sw
domestic = {"ya.ru", "vk.com"}
def _https(host, ip):
return _ok(ip) if host in domestic else ProbeResult(ip=ip, kind=KIND_CONNECT)
confirmed = self._run_all(_https, subnets=sw.SubnetVerdict(sw.CODE_CONFIRMED, 18, 20, 0, 15))
self.assertIn("Это режим «белых списков»: из российских сетей отвечают 18 из 20", confirmed["problems"][0]["text"])
# Зарубежные сети отвечают: закрыты именно контрольные сайты, «белых списков» нет.
refuted = self._run_all(_https, subnets=sw.SubnetVerdict(sw.CODE_REFUTED, 19, 20, 12, 15))
self.assertFalse(any("белых списков" in item["text"] for item in refuted["problems"]))
def test_subnets_are_not_checked_without_the_sign(self) -> None:
nets: list = []
self._run_all(lambda _host, ip: _ok(ip), net_out=nets)
self.assertEqual(nets[0].subnet_checks, 0)
def test_foreign_controls_open_is_not_whitelist_mode(self) -> None:
"""Один закрытый сайт при живых зарубежных контрольных — обычная блокировка."""
result = self._run_all(
lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "x.com" else _ok(ip)
)
self.assertFalse(any("белых списков" in item["text"] for item in result["problems"]))
self.assertTrue(any(item["target"] == "x.com" and item["action"] == "strategy" for item in result["problems"]))
def test_domestic_site_down_too_is_no_internet_not_whitelist(self) -> None:
result = self._run_all(lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT))
first = result["problems"][0]["text"]
self.assertIn("Не открываются даже контрольные сайты (Google, Cloudflare, Яндекс, ВКонтакте)", first)
self.assertNotIn("белых списков", first)
def test_working_sites_keep_service_order_and_offline_hides_site_noise(self) -> None:
# У YouTube подменён DNS (уровень «предупреждение»), но в «Открываются»
# он не должен выскакивать раньше Discord.
def _https(host, ip):
return _ok(ip)
from diagnostics.freeze_check import FreezeServer, FreezeState
from diagnostics.voice_check import VoiceServer
def _system(host):
if host == "www.youtube.com":
return DnsAnswer(ips=(), status=DNS_STATUS_NAME_ERROR)
return DnsAnswer(ips=DISCORD_REAL)
net = _Net(
system=_system,
https=_https,
voice=(VoiceServer("CF", "stun", True, "отвечает"),),
freeze=(FreezeServer("Akamai", FreezeState.OK, "получено 32 КБ"),),
)
result = net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
youtube = next(item for item in result["services"] if item["key"] == "youtube")
self.assertEqual(youtube["level"], "warn")
self.assertEqual(result["working"][:2], ["Discord", "YouTube"])
offline = self._run_all(lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT))
texts = [problem["text"] for problem in offline["problems"]]
self.assertIn("контрольные сайты", texts[0])
# Причина одна — нет интернета; строки «X не открывается» по каждому сайту не нужны.
self.assertEqual(len(texts), 1)
def test_controls_cut_by_deadline_do_not_hide_real_blocks(self) -> None:
controls = {"www.google.com", "www.cloudflare.com"}
def _https(host, ip):
if host in controls:
return ProbeResult(ip=ip, kind="cancelled")
if host == "x.com":
return ProbeResult(ip=ip, kind=KIND_RESET)
return _ok(ip)
result = self._run_all(_https)
texts = [problem["text"] for problem in result["problems"]]
self.assertFalse(any("контрольные сайты" in text for text in texts))
self.assertTrue(any(problem["target"] == "x.com" for problem in result["problems"]))
def test_strategy_button_only_for_blocks_a_strategy_can_bypass(self) -> None:
result = self._run_all(
lambda host, ip: ProbeResult(ip=ip, kind=KIND_CONNECT) if host == "x.com" else _ok(ip)
)
problem = next(item for item in result["problems"] if item["target"] == "x.com")
self.assertEqual(problem["action"], "")
def test_blocked_site_gets_strategy_action_with_its_host(self) -> None:
result = self._run_all(lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET) if host == "x.com" else _ok(ip))
problem = result["problems"][0]
self.assertEqual(problem["action"], "strategy")
self.assertEqual(problem["target"], "x.com")
self.assertIn("Discord", result["working"])
self.assertEqual(result["voice"]["level"], "ok")
self.assertEqual(result["freeze"]["level"], "ok")
class TlsProbeFailureTests(unittest.TestCase):
"""Классификация сбоев запроса «как Chrome» на настоящих сокетах (без TLS-сервера)."""
def _server(self, behaviour):
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.bind(("127.0.0.1", 0))
listener.listen(1)
port = listener.getsockname()[1]
def _serve():
try:
conn, _ = listener.accept()
behaviour(conn)
except OSError:
pass
finally:
listener.close()
threading.Thread(target=_serve, daemon=True).start()
return port
def _get(self, port, timeout=1.0):
return chrome_get("example.com", "127.0.0.1", timeout=timeout, port=port)
def test_reset_during_handshake_is_reset_or_tls(self) -> None:
def _rst(conn):
conn.recv(1024)
conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, b"\x01\x00\x00\x00\x00\x00\x00\x00")
conn.close()
result = self._get(self._server(_rst))
self.assertIn(result.kind, (KIND_RESET, KIND_TLS))
self.assertEqual(judge_reach(result), ReachState.DPI)
def test_silent_server_after_connect_is_timeout(self) -> None:
def _silent(conn):
time.sleep(2)
conn.close()
result = self._get(self._server(_silent), timeout=0.5)
self.assertEqual(result.kind, KIND_TIMEOUT)
self.assertEqual(result.stage, "tls")
def test_closed_port_is_connect_failure(self) -> None:
probe = socket.socket()
probe.bind(("127.0.0.1", 0))
port = probe.getsockname()[1]
probe.close()
result = self._get(port)
self.assertEqual(result.kind, KIND_CONNECT)
self.assertEqual(judge_reach(result), ReachState.IP_BLOCK)
class NoVideoNodeTests(unittest.TestCase):
def test_youtube_without_a_real_video_node_stays_open_and_the_video_is_unknown(self) -> None:
net = _Net()
net.video_hosts = ()
lines: list[str] = []
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
youtube = next(item for item in result["services"] if item["key"] == "youtube")
video = next(item for item in youtube["targets"] if item["purpose"] == "видео")
self.assertEqual((video["ok"], video["state"]), (False, "unknown"))
self.assertNotEqual(youtube["level"], "fail")
self.assertEqual(video["note"], "видеосервер не проверялся")
class BlockKindInReportTests(unittest.TestCase):
"""В отчёте у каждого сайта назван вид блокировки: по адресу, по имени или обрыв после 16 КБ."""
@staticmethod
def _stalls(received: int, requests: int):
vp = engine.volume_probe
return vp.VolumeRun(vp.RUN_STALLED, received, requests)
def test_cut_after_16kb_is_found_on_a_site_with_a_small_page(self) -> None:
"""Главная страница короткая и «открывается», но по соединению не проходит больше 14 КБ."""
vp = engine.volume_probe
lines: list[str] = []
net = _Net()
net.volume_facts = lambda host, _ip: (
vp.VolumeFacts(self._stalls(14_500, 20), self._stalls(14_100, 40))
if host == "discord.com"
else vp.VolumeFacts(vp.VolumeRun(vp.RUN_PASSED, 49_500, 4))
)
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
discord = next(item for item in result["services"] if item["key"] == "discord")
main = next(item for item in discord["targets"] if item["main"])
self.assertEqual((main["state"], main["ok"], main["kind"], main["volume"]), ("freeze", False, "cut16", "cut"))
self.assertEqual(main["kind_title"], "Обрыв после 16 КБ")
self.assertIn("соединение замирает после 14 КБ", main["short"])
self.assertEqual((discord["level"], discord["kind"]), ("fail", "cut16"))
self.assertTrue(discord["headline"].startswith("Discord грузится не до конца: загрузка обрывается после 16 КБ"))
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertEqual((problem["kind"], problem["title"], problem["action"]), ("cut16", "Discord", "strategy"))
self.assertNotIn("Discord", result["working"])
self.assertIn(" ℹ️ Вид блокировки: Обрыв после 16 КБ", lines)
def test_single_stall_does_not_turn_a_working_site_red(self) -> None:
vp = engine.volume_probe
net = _Net()
net.volume_facts = lambda _host, _ip: vp.VolumeFacts(
self._stalls(14_500, 20), vp.VolumeRun(vp.RUN_PASSED, 49_500, 70)
)
lines: list[str] = []
result = net.run(engine.run_blockcheck, "main", emit=lines.append)
self.assertEqual([item["level"] for item in result["services"]], ["ok", "ok"])
self.assertFalse([item for item in result["problems"] if item["level"] in ("fail", "warn")])
self.assertIn("Discord", result["working"])
self.assertIn("случайный сбой", "\n".join(lines))
def test_big_page_and_broken_site_are_not_probed_for_volume(self) -> None:
net = _Net(https=lambda host, ip: _ok(ip, body_size=40_000))
net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
# Страница больше окна обрыва пришла целиком — повторять незачем.
self.assertNotIn("discord.com", net.volume_asked)
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET, stage="tls"))
net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
self.assertEqual(net.volume_asked, [])
def test_control_sites_are_not_loaded_with_volume_requests(self) -> None:
net = _Net()
net.run(engine.run_blockcheck, "all", emit=lambda _line: None)
self.assertIn("discord.com", net.volume_asked)
for host in ("www.google.com", "www.cloudflare.com", "ya.ru", "vk.com"):
self.assertNotIn(host, net.volume_asked)
def test_block_by_name_goes_to_its_own_group(self) -> None:
def facts(host, result):
return engine.block_cause.CauseFacts(
host=host,
result=result,
neutral=engine.block_cause.HelloResult(engine.block_cause.HELLO_OK),
nameless=engine.block_cause.HelloResult(engine.block_cause.HELLO_RESET),
)
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET, stage="tls"), cause_facts=facts)
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["services"] if item["key"] == "discord")
self.assertEqual(discord["kind"], "sni")
self.assertIn("блокировка по имени сайта (SNI)", discord["headline"])
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertEqual((problem["kind"], problem["title"]), ("sni", "Discord"))
# Свидетельство стоит первым в советах и отдельно — чтобы экран не ставил перед ним стрелку.
self.assertEqual(problem["evidence"], problem["advice"][: len(problem["evidence"])])
self.assertTrue(problem["evidence"][0].startswith("Блокировка по имени сайта"))
def test_block_by_address_is_not_mixed_with_block_by_name(self) -> None:
def facts(host, result):
return engine.block_cause.CauseFacts(
host=host,
result=result,
# Тишина на любое имя, даже на разрешённое: так молчит фильтр, а не сам сервер.
neutral=engine.block_cause.HelloResult(engine.block_cause.HELLO_TIMEOUT),
nameless=engine.block_cause.HelloResult(engine.block_cause.HELLO_TIMEOUT),
allowed=engine.block_cause.HelloResult(engine.block_cause.HELLO_TIMEOUT),
)
net = _Net(https=lambda host, ip: ProbeResult(ip=ip, kind=KIND_RESET, stage="tls"), cause_facts=facts)
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
problem = next(item for item in result["problems"] if item["target"] == "discord.com")
self.assertEqual(problem["kind"], "ip")
self.assertIn("сервер заблокирован по адресу (IP)", problem["text"])
self.assertIn("Редакторе hosts", problem["advice"][-1])
def test_problems_about_the_network_have_their_own_kinds(self) -> None:
net = _Net(system=("0.0.0.0",))
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
kinds = {item["kind"] for item in result["problems"]}
self.assertIn("dns", kinds)
self.assertTrue(all(item["kind"] for item in result["problems"]))
class IcmpAddressTests(unittest.TestCase):
def test_icmp_address_keeps_network_byte_order_in_memory(self) -> None:
import socket
import struct
from utils.windows_icmp import _ipv4_to_dword
for ip in ("142.251.150.4", "162.159.137.232"):
with self.subTest(ip=ip):
self.assertEqual(struct.pack("<I", _ipv4_to_dword(ip)), socket.inet_aton(ip))
if __name__ == "__main__":
unittest.main()
class NoAddressGoesToDnsTests(unittest.TestCase):
def test_site_without_an_address_is_a_dns_problem_with_a_dns_button(self) -> None:
net = _Net(system=(), reference=())
result = net.run(engine.run_blockcheck, "main", emit=lambda _line: None)
discord = next(item for item in result["problems"] if item.get("site") == "discord")
self.assertEqual((discord["kind"], discord["action"]), ("dns", "dns"))
Reference in a new issue View git blame Copy permalink