www.tinc-vpn.org Git - tinc/blob - test/integration/device_multicast.py

   1 #!/usr/bin/env python3
   2
   3 """Test multicast device."""
   4
   5 import os
   6 import socket
   7 import struct
   8 import time
   9
  10 from testlib import check
  11 from testlib.log import log
  12 from testlib.proc import Tinc, Script
  13 from testlib.test import Test
  14
  15 MCAST_ADDR = "224.15.98.12"
  16 PORT = 38245
  17
  18
  19 def multicast_works() -> bool:
  20     """Check if multicast is supported and works."""
  21
  22     msg = b"foobar"
  23
  24     try:
  25         with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as server:
  26             server.bind((MCAST_ADDR, PORT))
  27
  28             req = struct.pack("=4sl", socket.inet_aton(MCAST_ADDR), socket.INADDR_ANY)
  29             server.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, req)
  30
  31             with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
  32                 client.sendto(msg, (MCAST_ADDR, PORT))
  33
  34             return msg == server.recv(16)
  35     except OSError:
  36         return False
  37
  38
  39 def test_no_mcast_support(foo: Tinc) -> None:
  40     """Check that startup fails on systems without multicast support."""
  41
  42     code = foo.tincd("-D").wait()
  43     check.failure(code)
  44     check.in_file(foo.sub("log"), f"Can't bind to {MCAST_ADDR}")
  45
  46
  47 def test_rx_tx(foo: Tinc) -> None:
  48     """Test sending real data to a multicast device."""
  49
  50     foo.start()
  51     packet = os.urandom(137)
  52
  53     with socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) as sock:
  54         for _ in range(5):
  55             sent = sock.sendto(packet, (MCAST_ADDR, PORT))
  56             log.info("sent broken packet (%d)", sent)
  57             time.sleep(0.1)
  58
  59     foo.add_script(Script.TINC_DOWN)
  60     foo.cmd("stop")
  61     foo[Script.TINC_DOWN].wait()
  62
  63     check.in_file(foo.sub("log"), "Read packet of 137 bytes from multicast socket")
  64
  65
  66 def test_device_multicast(ctx: Test) -> None:
  67     """Test multicast device."""
  68
  69     foo = ctx.node(init=True)
  70     foo.cmd("set", "DeviceType", "multicast")
  71
  72     log.info("check that multicast does not work without Device")
  73     _, err = foo.cmd("start", "-D", code=1)
  74     check.is_in("Device variable required for multicast socket", err)
  75
  76     log.info("check that Device requires a port")
  77     foo.cmd("set", "Device", "localhost")
  78     _, err = foo.cmd("start", "-D", code=1)
  79     check.is_in("Port number required", err)
  80
  81     log.info("check that multicast receives data")
  82     foo.cmd("set", "Device", f"{MCAST_ADDR} {PORT}")
  83     foo.cmd("set", "LogLevel", "10")
  84
  85     if multicast_works():
  86         test_rx_tx(foo)
  87     else:
  88         test_no_mcast_support(foo)
  89
  90
  91 with Test("test DeviceType = multicast") as context:
  92     test_device_multicast(context)