tests/topotests/lib/mcast-tester.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: ISC
   3 #
   4 # Copyright (C) 2021 by
   5 # Network Device Education Foundation, Inc. ("NetDEF")
   6
   7 """
   8 Subscribe to a multicast group so that the kernel sends an IGMP JOIN
   9 for the multicast group we subscribed to.
  10 """
  11
  12 import argparse
  13 import json
  14 import os
  15 import socket
  16 import struct
  17 import subprocess
  18 import sys
  19 import time
  20
  21
  22 #
  23 # Functions
  24 #
  25 def interface_name_to_index(name):
  26     "Gets the interface index using its name. Returns None on failure."
  27     interfaces = json.loads(subprocess.check_output("ip -j link show", shell=True))
  28
  29     for interface in interfaces:
  30         if interface["ifname"] == name:
  31             return interface["ifindex"]
  32
  33     return None
  34
  35
  36 def multicast_join(sock, ifindex, group, port):
  37     "Joins a multicast group."
  38     mreq = struct.pack(
  39         "=4sLL", socket.inet_aton(args.group), socket.INADDR_ANY, ifindex
  40     )
  41
  42     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
  43     sock.bind((group, port))
  44     sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
  45
  46
  47 #
  48 # Main code.
  49 #
  50 parser = argparse.ArgumentParser(description="Multicast RX utility")
  51 parser.add_argument("group", help="Multicast IP")
  52 parser.add_argument("interface", help="Interface name")
  53 parser.add_argument("--socket", help="Point to topotest UNIX socket")
  54 parser.add_argument(
  55     "--send", help="Transmit instead of join with interval", type=float, default=0
  56 )
  57 args = parser.parse_args()
  58
  59 ttl = 16
  60 port = 1000
  61
  62 # Get interface index/validate.
  63 ifindex = interface_name_to_index(args.interface)
  64 if ifindex is None:
  65     sys.stderr.write("Interface {} does not exists\n".format(args.interface))
  66     sys.exit(1)
  67
  68 # We need root privileges to set up multicast.
  69 if os.geteuid() != 0:
  70     sys.stderr.write("ERROR: You must have root privileges\n")
  71     sys.exit(1)
  72
  73 # Wait for topotest to synchronize with us.
  74 if not args.socket:
  75     toposock = None
  76 else:
  77     toposock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM, 0)
  78     while True:
  79         try:
  80             toposock.connect(args.socket)
  81             break
  82         except ConnectionRefusedError:
  83             time.sleep(1)
  84             continue
  85     # Set topotest socket non blocking so we can multiplex the main loop.
  86     toposock.setblocking(False)
  87
  88 msock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  89 if args.send > 0:
  90     # Prepare multicast bit in that interface.
  91     msock.setsockopt(
  92         socket.SOL_SOCKET,
  93         25,
  94         struct.pack("%ds" % len(args.interface), args.interface.encode("utf-8")),
  95     )
  96     # Set packets TTL.
  97     msock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, struct.pack("b", ttl))
  98     # Block to ensure packet send.
  99     msock.setblocking(True)
 100 else:
 101     multicast_join(msock, ifindex, args.group, port)
 102
 103
 104 def should_exit():
 105     if not toposock:
 106         # If we are sending then we have slept
 107         if not args.send:
 108             time.sleep(100)
 109         return False
 110     else:
 111         try:
 112             data = toposock.recv(1)
 113             if data == b"":
 114                 print(" -> Connection closed")
 115                 return True
 116         except BlockingIOError:
 117             return False
 118
 119
 120 counter = 0
 121 while not should_exit():
 122     if args.send > 0:
 123         msock.sendto(b"test %d" % counter, (args.group, port))
 124         counter += 1
 125         time.sleep(args.send)
 126
 127 msock.close()
 128 sys.exit(0)