test(#21): интеграционные тесты netns-стенда и документация харнесса
- test_netns_stand.py — 8 тестов: топология и линки, policy routing по fwmark, правила меток и kill-switch, готовность мока и детектора, доставка handshake из LAN и из шлюза с ростом счётчика метки, имитация недоступности провайдера (mode=drop), положительный контроль утечки и блокировка при kill-switch. Тесты пропускаются вне run.sh, чтобы не падать при случайном pytest-прогоне; - README.md харнесса — схема топологии, состав файлов, таблица тестов, переменные окружения и pitfalls (root/CAP_NET_ADMIN, rp_filter, отсутствие nft и ping в alpine-образе). Refs #21 Refs #15
This commit is contained in:
@@ -0,0 +1,210 @@
|
||||
"""Интеграционные тесты netns-стенда awg-dash (задача #21, эпик #15).
|
||||
|
||||
Проверяют, что стенд реально поднимает топологию client/gw/uplink, мок
|
||||
WireGuard-endpoint отвечает без настоящего провайдера, тестовый трафик
|
||||
маркируется, а kill-switch не пускает трафик напрямую.
|
||||
|
||||
Запускать только через `tests/netns/run.sh` (или `make test-netns`): тестам
|
||||
нужны root, поднятая топология и запущенные процессы стенда. Вне run.sh тесты
|
||||
помечаются skipped, чтобы случайный прогон pytest/CI не падал.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import time
|
||||
import unittest
|
||||
|
||||
import netns_common as nc
|
||||
|
||||
REASON = "netns-стенд не запущен: используйте tests/netns/run.sh (нужны root и CAP_NET_ADMIN)"
|
||||
KILLSWITCH_SCRIPT = str(nc.HERE / "topology.sh")
|
||||
|
||||
|
||||
@unittest.skipUnless(os.environ.get("AWG_NETNS_RUN") == "1", REASON)
|
||||
class NetnsStandTest(unittest.TestCase):
|
||||
"""Порядок методов значим: стенд stateful, шаги идут как test_01..test_08."""
|
||||
|
||||
maxDiff = None
|
||||
|
||||
def _killswitch(self, state: str) -> None:
|
||||
nc.sh(["bash", KILLSWITCH_SCRIPT, "killswitch", state])
|
||||
|
||||
# --- AC-3, шаг 1: топология ---------------------------------------------
|
||||
def test_01_namespaces_and_links(self) -> None:
|
||||
for ns in (nc.CLIENT_NS, nc.GW_NS, nc.UPLINK_NS):
|
||||
self.assertTrue(nc.ns_exists(ns), f"netns {ns} не создан")
|
||||
|
||||
self.assertIn(nc.CLIENT_IP, nc.addrs(nc.CLIENT_NS), "у клиента нет ожидаемого адреса")
|
||||
gw_addrs = nc.addrs(nc.GW_NS)
|
||||
self.assertIn(nc.GW_LAN_IP, gw_addrs, "у шлюза нет LAN-адреса")
|
||||
self.assertIn(nc.GW_WAN_IP, gw_addrs, "у шлюза нет WAN-адреса")
|
||||
self.assertIn(nc.UPLINK_IP, nc.addrs(nc.UPLINK_NS), "у аплинка нет ожидаемого адреса")
|
||||
|
||||
for ns, ifname in (
|
||||
(nc.CLIENT_NS, nc.VETH_CLIENT),
|
||||
(nc.GW_NS, nc.VETH_GW_LAN),
|
||||
(nc.GW_NS, nc.VETH_GW_WAN),
|
||||
(nc.UPLINK_NS, nc.VETH_UPLINK),
|
||||
):
|
||||
self.assertTrue(nc.link_is_up(ns, ifname), f"{ifname} в {ns} не UP")
|
||||
|
||||
forwarding = nc.ns_sh(nc.GW_NS, ["sysctl", "-n", "net.ipv4.ip_forward"]).stdout.strip()
|
||||
self.assertEqual("1", forwarding, "в шлюзе не включён ip_forward — транзит не пойдёт")
|
||||
|
||||
# --- policy routing без реального провайдера ------------------------------
|
||||
def test_02_policy_routing_by_fwmark(self) -> None:
|
||||
rules = nc.ns_sh(nc.GW_NS, ["ip", "rule", "show"]).stdout
|
||||
self.assertIn(f"lookup {nc.ROUTE_TABLE}", rules, "нет ip rule fwmark -> table")
|
||||
self.assertIn("fwmark", rules, "правило policy routing не содержит fwmark")
|
||||
|
||||
table = nc.ns_sh(nc.GW_NS, ["ip", "route", "show", "table", str(nc.ROUTE_TABLE)]).stdout
|
||||
self.assertIn(f"via {nc.UPLINK_IP}", table, f"в таблице {nc.ROUTE_TABLE} нет default via аплинк")
|
||||
|
||||
route_get = nc.ns_sh(
|
||||
nc.GW_NS, ["ip", "route", "get", nc.UPLINK_IP, "mark", str(nc.FWMARK)]
|
||||
).stdout
|
||||
self.assertIn(
|
||||
f"table {nc.ROUTE_TABLE}",
|
||||
route_get,
|
||||
f"маршрут для fwmark {nc.FWMARK} не берётся из таблицы {nc.ROUTE_TABLE}: {route_get}",
|
||||
)
|
||||
|
||||
def test_03_netfilter_mark_and_killswitch_rules(self) -> None:
|
||||
kind = nc.fw_kind()
|
||||
self.assertIn(kind, ("nft", "iptables"), "в netns шлюза нет ни nft, ни iptables")
|
||||
|
||||
if kind == "nft":
|
||||
comments = {str(rule.get("comment", "")) for rule in nc.nft_rules()}
|
||||
for needle in (nc.C_MARK_OUT, nc.C_MARK_FWD, nc.C_DROP):
|
||||
self.assertIn(needle, comments, f"нет правила nft с комментарием {needle}")
|
||||
else:
|
||||
rules = nc.iptables_rules()
|
||||
self.assertIn(nc.IPT_CHAIN_MARK, rules, "нет цепочки метки iptables")
|
||||
self.assertIn(nc.IPT_CHAIN_KS, rules, "нет цепочки kill-switch iptables")
|
||||
self.assertIn("MARK", rules, "нет правила MARK --set-mark")
|
||||
self.assertIn("DROP", rules, "нет правила DROP kill-switch")
|
||||
|
||||
# --- мок-endpoint --------------------------------------------------------
|
||||
def test_04_mock_endpoint_and_leak_sink_ready(self) -> None:
|
||||
self.assertTrue(nc.MOCK_READY.exists(), f"нет флага готовности мока {nc.MOCK_READY}")
|
||||
self.assertTrue(nc.LEAK_READY.exists(), f"нет флага готовности детектора {nc.LEAK_READY}")
|
||||
|
||||
listen = nc.wait_jsonl(nc.MOCK_LOG, lambda e: e.get("event") == "listen", timeout=3)
|
||||
self.assertIsNotNone(listen, "в логе мока нет события listen")
|
||||
self.assertIn(str(nc.MOCK_PORT), str(listen.get("addr")), "мок слушает не тот порт")
|
||||
|
||||
sink = nc.wait_jsonl(nc.LEAK_LOG, lambda e: e.get("event") == "listen", timeout=3)
|
||||
self.assertIsNotNone(sink, "детектор утечки не залогировал listen")
|
||||
|
||||
# --- AC-3, шаг 2: трафик маркируется и доходит до мока --------------------
|
||||
def test_05_forwarded_handshake_reaches_mock_and_is_marked(self) -> None:
|
||||
before = nc.fw_counters()
|
||||
|
||||
result = nc.probe(nc.CLIENT_NS, nc.UPLINK_IP, nc.MOCK_PORT, timeout=2.0)
|
||||
self.assertTrue(
|
||||
result.get("reply_received"),
|
||||
f"мок-endpoint не ответил на handshake initiation: {result}",
|
||||
)
|
||||
|
||||
rx = nc.wait_jsonl(
|
||||
nc.MOCK_LOG,
|
||||
lambda e: e.get("event") == "rx" and e.get("msg_type") == 1 and str(e.get("src", "")).startswith(nc.CLIENT_IP),
|
||||
timeout=3,
|
||||
)
|
||||
self.assertIsNotNone(rx, "мок не увидел handshake initiation от клиента")
|
||||
self.assertEqual(148, rx["length"], "длина handshake initiation не 148 байт")
|
||||
self.assertEqual("handshake_initiation", rx["msg_name"])
|
||||
|
||||
tx = nc.wait_jsonl(
|
||||
nc.MOCK_LOG,
|
||||
lambda e: e.get("event") == "tx" and e.get("msg_type") == 2,
|
||||
timeout=3,
|
||||
)
|
||||
self.assertIsNotNone(tx, "мок не отправил handshake response")
|
||||
self.assertEqual(92, tx["length"], "длина handshake response не 92 байта")
|
||||
|
||||
reply = result.get("reply") or {}
|
||||
self.assertEqual(
|
||||
reply.get("sender_index_echo"),
|
||||
reply.get("receiver_index"),
|
||||
"мок не подставил sender_index клиента как receiver_index",
|
||||
)
|
||||
|
||||
after = nc.fw_counters()
|
||||
self.assertIsNotNone(after["mark_fwd"], "нет счётчика правила метки транзитного трафика")
|
||||
self.assertGreater(
|
||||
after["mark_fwd"], before["mark_fwd"] or 0, "счётчик правила метки (FORWARD) не вырос — трафик не маркируется"
|
||||
)
|
||||
|
||||
def test_06_gateway_originated_handshake_is_marked(self) -> None:
|
||||
before = nc.fw_counters()
|
||||
result = nc.probe(nc.GW_NS, nc.UPLINK_IP, nc.MOCK_PORT, timeout=2.0)
|
||||
self.assertTrue(result.get("reply_received"), f"мок не ответил на handshake из netns шлюза: {result}")
|
||||
|
||||
after = nc.fw_counters()
|
||||
self.assertIsNotNone(after["mark_out"], "нет счётчика правила метки исходящего трафика шлюза")
|
||||
self.assertGreater(
|
||||
after["mark_out"], before["mark_out"] or 0, "счётчик правила метки (OUTPUT) не вырос"
|
||||
)
|
||||
|
||||
# --- имитация недоступности провайдера ------------------------------------
|
||||
def test_07_provider_outage_drop_mode(self) -> None:
|
||||
nc.set_mock_mode("drop")
|
||||
try:
|
||||
result = nc.probe(nc.CLIENT_NS, nc.UPLINK_IP, nc.MOCK_PORT, expect_timeout=True, timeout=1.5)
|
||||
self.assertTrue(result.get("ok"), f"в режиме drop ожидалось отсутствие ответа: {result}")
|
||||
self.assertFalse(result.get("reply_received"), "мок ответил, хотя провайдер «недоступен»")
|
||||
|
||||
rx = nc.wait_jsonl(
|
||||
nc.MOCK_LOG,
|
||||
lambda e: e.get("event") == "rx" and e.get("mode") == "drop",
|
||||
timeout=3,
|
||||
)
|
||||
self.assertIsNotNone(rx, "мок в режиме drop не зафиксировал пакет")
|
||||
finally:
|
||||
nc.set_mock_mode("respond")
|
||||
|
||||
recovered = nc.probe(nc.CLIENT_NS, nc.UPLINK_IP, nc.MOCK_PORT, timeout=2.0)
|
||||
self.assertTrue(recovered.get("reply_received"), "стенд не восстановился после возврата режима respond")
|
||||
|
||||
# --- kill-switch (fail-closed) -------------------------------------------
|
||||
def test_08_killswitch_blocks_direct_egress(self) -> None:
|
||||
marker = "awgtest-leak-probe"
|
||||
|
||||
# 1) положительный контроль: при выключенном kill-switch трафик доходит
|
||||
self._killswitch("off")
|
||||
nc.probe(nc.CLIENT_NS, nc.LEAK_IP, nc.LEAK_PORT, raw=marker, expect_timeout=True, timeout=1.0)
|
||||
control = nc.wait_jsonl(
|
||||
nc.LEAK_LOG,
|
||||
lambda e: e.get("event") == "leak" and e.get("length") == len(marker),
|
||||
timeout=3,
|
||||
)
|
||||
self.assertIsNotNone(
|
||||
control,
|
||||
"контрольная проверка не прошла: при выключенном kill-switch трафик не дошёл до детектора",
|
||||
)
|
||||
baseline = len(nc.leak_events())
|
||||
|
||||
# 2) kill-switch включён -> трафик, предназначенный туннелю, не должен утечь
|
||||
self._killswitch("on")
|
||||
before_drop = nc.fw_counters()
|
||||
nc.probe(nc.CLIENT_NS, nc.LEAK_IP, nc.LEAK_PORT, raw=marker, expect_timeout=True, timeout=1.0)
|
||||
time.sleep(1.0)
|
||||
|
||||
self.assertEqual(
|
||||
baseline,
|
||||
len(nc.leak_events()),
|
||||
"трафик утёк напрямую при включённом kill-switch (fail-closed нарушен)",
|
||||
)
|
||||
after_drop = nc.fw_counters()
|
||||
if after_drop["killswitch_drop"] is not None:
|
||||
self.assertGreater(
|
||||
after_drop["killswitch_drop"],
|
||||
before_drop["killswitch_drop"] or 0,
|
||||
"счётчик DROP kill-switch не вырос — правило не сработало",
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user