diff --git a/tests/netns/leak_sink.py b/tests/netns/leak_sink.py new file mode 100644 index 0000000..bd2b752 --- /dev/null +++ b/tests/netns/leak_sink.py @@ -0,0 +1,120 @@ +#!/usr/bin/env python3 +"""leak_sink.py — детектор утечки трафика для netns-стенда (задача #21). + +Слушает UDP и TCP порт 9999 на «публичном» адресе 198.51.100.9 внутри netns +`uplink` (адрес из TEST-NET-2, RFC 5737). Если при включённом kill-switch +шлюза сюда пришёл хоть один пакет — значит трафик, предназначенный для +туннеля, утёк напрямую: тест падает. + +Лог — JSONL, пишется немедленно (тесты читают файл, не дожидаясь остановки). +""" + +from __future__ import annotations + +import argparse +import json +import os +import socket +import sys +import threading +import time + + +class JsonlLog: + def __init__(self, path: str | None) -> None: + self._lock = threading.Lock() + self._fh = None + if path: + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + self._fh = open(path, "a", buffering=1, encoding="utf-8") + + def write(self, **event: object) -> None: + if self._fh is None: + return + event.setdefault("ts", round(time.time(), 6)) + with self._lock: + self._fh.write(json.dumps(event, sort_keys=True) + "\n") + self._fh.flush() + + def close(self) -> None: + if self._fh is not None: + self._fh.close() + self._fh = None + + +def udp_loop(bind: tuple[str, int], log: JsonlLog, stop: threading.Event) -> None: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.bind(bind) + sock.settimeout(0.25) + while not stop.is_set(): + try: + payload, peer = sock.recvfrom(65535) + except socket.timeout: + continue + except OSError: + break + log.write(event="leak", proto="udp", length=len(payload), src=f"{peer[0]}:{peer[1]}") + sock.close() + + +def tcp_loop(bind: tuple[str, int], log: JsonlLog, stop: threading.Event) -> None: + srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + srv.bind(bind) + srv.listen(8) + srv.settimeout(0.25) + while not stop.is_set(): + try: + conn, peer = srv.accept() + except socket.timeout: + continue + except OSError: + break + log.write(event="leak", proto="tcp", src=f"{peer[0]}:{peer[1]}") + conn.close() + srv.close() + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description="Детектор утечки трафика для netns-стенда (#21)") + ap.add_argument("--ip", default=os.environ.get("AWG_LEAK_IP", "198.51.100.9")) + ap.add_argument("--port", type=int, default=int(os.environ.get("AWG_LEAK_PORT", "9999"))) + ap.add_argument("--log", default=os.environ.get("AWG_LEAK_LOG", "")) + ap.add_argument("--ready", default=os.environ.get("AWG_LEAK_READY", "")) + args = ap.parse_args(argv) + + bind = (args.ip, args.port) + log = JsonlLog(args.log or None) + stop = threading.Event() + + udp = threading.Thread(target=udp_loop, args=(bind, log, stop), daemon=True) + tcp = threading.Thread(target=tcp_loop, args=(bind, log, stop), daemon=True) + udp.start() + tcp.start() + + if args.ready: + os.makedirs(os.path.dirname(os.path.abspath(args.ready)) or ".", exist_ok=True) + with open(args.ready, "w", encoding="utf-8") as fh: + fh.write(f"{os.getpid()}\n") + + log.write(event="listen", addr=f"{args.ip}:{args.port}", pid=os.getpid()) + print(f"leak-sink: udp+tcp {args.ip}:{args.port} listen", flush=True) + try: + while True: + time.sleep(0.5) + except KeyboardInterrupt: + pass + finally: + stop.set() + udp.join(timeout=2) + tcp.join(timeout=2) + log.write(event="stop") + log.close() + if args.ready and os.path.exists(args.ready): + os.unlink(args.ready) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/netns/mock_wg_endpoint.py b/tests/netns/mock_wg_endpoint.py new file mode 100644 index 0000000..deb1ead --- /dev/null +++ b/tests/netns/mock_wg_endpoint.py @@ -0,0 +1,210 @@ +#!/usr/bin/env python3 +"""mock_wg_endpoint.py — мок WireGuard/AmneziaWG-endpoint для netns-стенда (задача #21). + +Зачем: интеграционные тесты split-tunnel должны идти без реального VPN-провайдера и +без настоящего туннеля. Мок слушает UDP-порт 51820 в netns `uplink` («ISP»), +распознаёт тип сообщения WireGuard по первому байту и отвечает на handshake +initiation (type 1, 148 байт) корректным по структуре handshake response +(type 2, 92 байта), подставляя sender_index клиента как receiver_index. + +ЧЕГО МОК НЕ ДЕЛАЕТ (важно, чтобы не принять его за настоящий peer): + * не выполняет Noise IK handshake и никакой криптографии — поля «зашифрованы» + случайными байтами; + * не маршрутизирует трафик и не поднимает TUN-интерфейс. +Поэтому мок НЕ подходит для проверки шифрования/MTU/throughput — только для +проверки топологии, policy routing, метки трафика и fail-closed сценариев. + +Режим работы переключается на лету через файл `--mode-file`: + respond — отвечать на handshake initiation (провайдер доступен); + drop — молча считать пакеты и не отвечать (имитация «провайдер недоступен»). + +Лог: JSONL (по строке на событие), пишется и сбрасывается на диск сразу, чтобы +тесты могли его читать без гонок. Файл `--ready` создаётся после bind(). + +Только stdlib — запускается и на хосте, и в alpine-образе приложения. +""" + +from __future__ import annotations + +import argparse +import json +import os +import random +import socket +import struct +import sys +import time + +MSG_HANDSHAKE_INITIATION = 1 +MSG_HANDSHAKE_RESPONSE = 2 +MSG_COOKIE_REPLY = 3 +MSG_TRANSPORT_DATA = 4 + +TYPE_NAMES = { + MSG_HANDSHAKE_INITIATION: "handshake_initiation", + MSG_HANDSHAKE_RESPONSE: "handshake_response", + MSG_COOKIE_REPLY: "cookie_reply", + MSG_TRANSPORT_DATA: "transport_data", +} + +HANDSHAKE_INITIATION_LEN = 148 +HANDSHAKE_RESPONSE_LEN = 92 + + +def msg_type(payload: bytes) -> int: + """WireGuard message type: младший байт little-endian uint32 (3 байта reserved).""" + if len(payload) < 4: + return -1 + return payload[0] + + +def build_handshake_response(initiation: bytes) -> bytes: + """Собрать структурно корректный type 2 (92 байта) в ответ на type 1. + + Раскладка: type(1)+reserved(3) sender(4) receiver(4) ephemeral(32) + encrypted_nothing(16) mac1(16) mac2(16) = 92 + """ + sender_index = initiation[4:8] if len(initiation) >= 8 else b"\x00" * 4 + out = bytearray(HANDSHAKE_RESPONSE_LEN) + out[0] = MSG_HANDSHAKE_RESPONSE + # sender_index мока — случайный, receiver_index = sender_index клиента + out[4:8] = struct.pack(" None: + self._fh = None + if path: + os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True) + self._fh = open(path, "a", buffering=1, encoding="utf-8") + + def write(self, **event: object) -> None: + if self._fh is None: + return + event.setdefault("ts", round(time.time(), 6)) + self._fh.write(json.dumps(event, sort_keys=True) + "\n") + self._fh.flush() + + def close(self) -> None: + if self._fh is not None: + self._fh.close() + self._fh = None + + +def read_mode(mode_file: str | None, default: str) -> str: + """Режим читается из файла на каждом пакете: тесты переключают без рестарта.""" + if not mode_file: + return default + try: + with open(mode_file, encoding="utf-8") as fh: + mode = fh.read().strip() + except OSError: + return default + return mode if mode in ("respond", "drop") else default + + +def main(argv: list[str] | None = None) -> int: + ap = argparse.ArgumentParser(description="Мок WireGuard-endpoint для netns-стенда (#21)") + ap.add_argument("--bind", default="0.0.0.0:51820", help="адрес:порт для прослушивания UDP") + ap.add_argument("--log", default=os.environ.get("AWG_MOCK_LOG", ""), help="JSONL-лог событий") + ap.add_argument("--ready", default=os.environ.get("AWG_MOCK_READY", ""), help="файл-флаг готовности") + ap.add_argument( + "--mode-file", + default=os.environ.get("AWG_MOCK_MODE_FILE", ""), + help="файл с режимом respond|drop (переключение на лету)", + ) + ap.add_argument("--mode", default=os.environ.get("AWG_MOCK_MODE", "respond"), choices=["respond", "drop"]) + args = ap.parse_args(argv) + + host, _, port_s = args.bind.rpartition(":") + host = host or "0.0.0.0" + port = int(port_s or "51820") + + log = JsonlLog(args.log or None) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind((host, port)) + except OSError as exc: + log.write(event="bind_error", addr=f"{host}:{port}", error=str(exc)) + print(f"mock-endpoint: bind {host}:{port} failed: {exc}", file=sys.stderr) + return 4 + sock.settimeout(0.25) + + if args.ready: + os.makedirs(os.path.dirname(os.path.abspath(args.ready)) or ".", exist_ok=True) + with open(args.ready, "w", encoding="utf-8") as fh: + fh.write(f"{os.getpid()}\n") + + log.write(event="listen", addr=f"{host}:{port}", pid=os.getpid(), mode=args.mode) + print(f"mock-endpoint: udp/{port} listen, mode={args.mode}", flush=True) + + rx_total = 0 + tx_total = 0 + try: + while True: + try: + payload, peer = sock.recvfrom(65535) + except socket.timeout: + continue + except OSError: + break + + rx_total += 1 + mtype = msg_type(payload) + mode = read_mode(args.mode_file or None, args.mode) + log.write( + event="rx", + msg_type=mtype, + msg_name=TYPE_NAMES.get(mtype, "unknown"), + length=len(payload), + src=f"{peer[0]}:{peer[1]}", + rx_total=rx_total, + mode=mode, + ) + + if mode == "drop": + continue + if mtype != MSG_HANDSHAKE_INITIATION: + log.write(event="ignored", reason="not_handshake_initiation", msg_type=mtype) + continue + if len(payload) != HANDSHAKE_INITIATION_LEN: + log.write( + event="ignored", + reason="bad_initiation_length", + length=len(payload), + expected=HANDSHAKE_INITIATION_LEN, + ) + continue + + reply = build_handshake_response(payload) + sock.sendto(reply, peer) + tx_total += 1 + log.write( + event="tx", + msg_type=MSG_HANDSHAKE_RESPONSE, + length=len(reply), + dst=f"{peer[0]}:{peer[1]}", + tx_total=tx_total, + ) + except KeyboardInterrupt: + pass + finally: + sock.close() + log.write(event="stop", rx_total=rx_total, tx_total=tx_total) + log.close() + if args.ready and os.path.exists(args.ready): + os.unlink(args.ready) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/netns/probe.py b/tests/netns/probe.py new file mode 100644 index 0000000..4ff0fe8 --- /dev/null +++ b/tests/netns/probe.py @@ -0,0 +1,113 @@ +#!/usr/bin/env python3 +"""probe.py — генератор тестового трафика из netns-стенда (задача #21). + +Запускается внутри конкретного netns (`ip netns exec python3 probe.py ...`) +и умеет: + + * `--wg-initiation` (по умолчанию) — отправить WireGuard handshake initiation + (type 1, 148 байт, раскладка как в протоколе) и дождаться ответа мока; + * `--raw TEXT` — отправить произвольный UDP-datagram (для проверок утечки / + kill-switch), по умолчанию ответа не ожидает. + +Результат печатается одной JSON-строкой в stdout, код возврата: + 0 — ожидаемое поведение достигнуто, 1 — нет, 2 — ошибка аргументов/сокета. +""" + +from __future__ import annotations + +import argparse +import json +import os +import struct +import sys +import time + +MSG_HANDSHAKE_INITIATION = 1 +MSG_HANDSHAKE_RESPONSE = 2 +HANDSHAKE_INITIATION_LEN = 148 + + +def build_handshake_initiation(sender_index: int | None = None) -> bytes: + """148 байт: type(1)+reserved(3) sender(4) ephemeral(32) enc_static(48) enc_ts(28) mac1(16) mac2(16).""" + if sender_index is None: + sender_index = struct.unpack(" int: + ap = argparse.ArgumentParser(description="Тестовый трафик для netns-стенда (#21)") + ap.add_argument("--dst", required=True, help="IP назначения") + ap.add_argument("--port", type=int, required=True, help="UDP-порт назначения") + ap.add_argument("--timeout", type=float, default=2.0, help="ожидание ответа, с") + ap.add_argument("--raw", default=None, help="отправить произвольный текст вместо handshake initiation") + ap.add_argument("--expect-timeout", action="store_true", help="ожидать отсутствия ответа (для kill-switch/drop)") + ap.add_argument("--source-port", type=int, default=0, help="локальный порт (0 — случайный)") + args = ap.parse_args(argv) + + if args.raw is None: + payload = build_handshake_initiation() + expect_reply = not args.expect_timeout + kind = "wg_handshake_initiation" + else: + payload = args.raw.encode("utf-8") + # для raw-трафика ответ не ожидается, если явно не сказано обратное + expect_reply = not args.expect_timeout + kind = "raw_udp" + + import socket + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + if args.source_port: + sock.bind(("", args.source_port)) + sock.settimeout(args.timeout) + + started = time.time() + try: + sent = sock.sendto(payload, (args.dst, args.port)) + except OSError as exc: + print(json.dumps({"ok": False, "kind": kind, "error": f"sendto: {exc}"})) + return 2 + + reply = None + try: + data, peer = sock.recvfrom(65535) + reply = { + "type": data[0] if data else -1, + "length": len(data or b""), + "src": f"{peer[0]}:{peer[1]}", + "receiver_index": ( + struct.unpack("= 12 and data[0] == MSG_HANDSHAKE_RESPONSE else None + ), + "sender_index_echo": struct.unpack("