test(#21): мок WireGuard-endpoint, детектор утечки и генератор трафика

- mock_wg_endpoint.py — UDP-эндпоинт, повторяющий раскладку сообщений
  WireGuard (initiation 148 байт -> response 92 байта), режимы respond/drop
  переключаются на лету через файл; JSONL-лог для ассертов;
- leak_sink.py — UDP+TCP детектор утечки на 198.51.100.9:9999 внутри netns
  uplink (RFC 5737 TEST-NET-2);
- probe.py — отправка handshake initiation или raw-UDP из указанного netns
  с разбором ответа в JSON. Только stdlib: работает в alpine-образе.

Refs #21
Refs #15

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())

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("<I", random.getrandbits(32))
+    out[8:12] = sender_index
+    out[12:44] = os.urandom(32)   # unencrypted_ephemeral
+    out[44:60] = os.urandom(16)   # encrypted_nothing + poly1305 tag
+    out[60:76] = os.urandom(16)   # mac1
+    out[76:92] = os.urandom(16)   # mac2
+    return bytes(out)
+
+
+class JsonlLog:
+    """Пишущий JSONL-лог с немедленным flush (тесты читают файл параллельно)."""
+
+    def __init__(self, path: str | None) -> 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())

tests/netns/probe.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""probe.py — генератор тестового трафика из netns-стенда (задача #21).
+
+Запускается внутри конкретного netns (`ip netns exec <ns> 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("<I", os.urandom(4))[0]
+    out = bytearray(HANDSHAKE_INITIATION_LEN)
+    out[0] = MSG_HANDSHAKE_INITIATION
+    out[4:8] = struct.pack("<I", sender_index)
+    out[8:40] = os.urandom(32)    # unencrypted_ephemeral
+    out[40:88] = os.urandom(48)   # encrypted_static (32) + poly1305 tag (16)
+    out[88:116] = os.urandom(28)  # encrypted_timestamp (12) + tag (16)
+    out[116:132] = os.urandom(16)  # mac1
+    out[132:148] = os.urandom(16)  # mac2
+    return bytes(out)
+
+
+def main(argv: list[str] | None = None) -> 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("<I", data[8:12])[0] if len(data) >= 12 and data[0] == MSG_HANDSHAKE_RESPONSE else None
+            ),
+            "sender_index_echo": struct.unpack("<I", payload[4:8])[0],
+        }
+    except socket.timeout:
+        reply = None
+    finally:
+        sock.close()
+
+    got_reply = reply is not None
+    ok = got_reply if expect_reply else (not got_reply)
+    out = {
+        "ok": ok,
+        "kind": kind,
+        "sent": sent,
+        "dst": f"{args.dst}:{args.port}",
+        "expect_reply": expect_reply,
+        "reply_received": got_reply,
+        "reply": reply,
+        "rtt_ms": round((time.time() - started) * 1000, 2),
+    }
+    print(json.dumps(out, sort_keys=True))
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())