#! /usr/bin/env python3 # SPDX-License-Identifier: GPL-2.0 import argparse import ctypes import errno import hashlib import os import select import signal import socket import subprocess import sys import atexit from pwd import getpwuid from os import stat # Allow utils module to be imported from different directory this_dir = os.path.dirname(os.path.realpath(__file__)) sys.path.append(os.path.join(this_dir, "../")) from lib.py.utils import ip libc = ctypes.cdll.LoadLibrary('libc.so.6') setns = libc.setns net0 = 'net0' net1 = 'net1' veth0 = 'veth0' veth1 = 'veth1' # Helper function for creating a socket inside a network namespace. # We need this because otherwise RDS will detect that the two TCP # sockets are on the same interface and use the loop transport instead # of the TCP transport. def netns_socket(netns, *args): u0, u1 = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) child = os.fork() if child == 0: # change network namespace with open(f'/var/run/netns/{netns}') as f: try: ret = setns(f.fileno(), 0) except IOError as e: print(e.errno) print(e) # create socket in target namespace s = socket.socket(*args) # send resulting socket to parent socket.send_fds(u0, [], [s.fileno()]) sys.exit(0) # receive socket from child _, s, _, _ = socket.recv_fds(u1, 0, 1) os.waitpid(child, 0) u0.close() u1.close() return socket.fromfd(s[0], *args) def signal_handler(sig, frame): print('Test timed out') sys.exit(1) #Parse out command line arguments. We take an optional # timeout parameter and an optional log output folder parser = argparse.ArgumentParser(description="init script args", formatter_class=argparse.ArgumentDefaultsHelpFormatter) parser.add_argument("-d", "--logdir", action="store", help="directory to store logs", default="/tmp") parser.add_argument('--timeout', help="timeout to terminate hung test", type=int, default=0) parser.add_argument('-l', '--loss', help="Simulate tcp packet loss", type=int, default=0) parser.add_argument('-c', '--corruption', help="Simulate tcp packet corruption", type=int, default=0) parser.add_argument('-u', '--duplicate', help="Simulate tcp packet duplication", type=int, default=0) args = parser.parse_args() logdir=args.logdir packet_loss=str(args.loss)+'%' packet_corruption=str(args.corruption)+'%' packet_duplicate=str(args.duplicate)+'%' ip(f"netns add {net0}") ip(f"netns add {net1}") ip(f"link add type veth") addrs = [ # we technically don't need different port numbers, but this will # help identify traffic in the network analyzer ('10.0.0.1', 10000), ('10.0.0.2', 20000), ] # move interfaces to separate namespaces so they can no longer be # bound directly; this prevents rds from switching over from the tcp # transport to the loop transport. ip(f"link set {veth0} netns {net0} up") ip(f"link set {veth1} netns {net1} up") # add addresses ip(f"-n {net0} addr add {addrs[0][0]}/32 dev {veth0}") ip(f"-n {net1} addr add {addrs[1][0]}/32 dev {veth1}") # add routes ip(f"-n {net0} route add {addrs[1][0]}/32 dev {veth0}") ip(f"-n {net1} route add {addrs[0][0]}/32 dev {veth1}") # sanity check that our two interfaces/addresses are correctly set up # and communicating by doing a single ping ip(f"netns exec {net0} ping -c 1 {addrs[1][0]}") # Start a packet capture on each network for net in [net0, net1]: tcpdump_pid = os.fork() if tcpdump_pid == 0: pcap = logdir+'/'+net+'.pcap' subprocess.check_call(['touch', pcap]) user = getpwuid(stat(pcap).st_uid).pw_name ip(f"netns exec {net} /usr/sbin/tcpdump -Z {user} -i any -w {pcap}") sys.exit(0) # simulate packet loss, duplication and corruption for net, iface in [(net0, veth0), (net1, veth1)]: ip(f"netns exec {net} /usr/sbin/tc qdisc add dev {iface} root netem \ corrupt {packet_corruption} loss {packet_loss} duplicate \ {packet_duplicate}") # add a timeout if args.timeout > 0: signal.alarm(args.timeout) signal.signal(signal.SIGALRM, signal_handler) sockets = [ netns_socket(net0, socket.AF_RDS, socket.SOCK_SEQPACKET), netns_socket(net1, socket.AF_RDS, socket.SOCK_SEQPACKET), ] for s, addr in zip(sockets, addrs): s.bind(addr) s.setblocking(0) fileno_to_socket = { s.fileno(): s for s in sockets } addr_to_socket = { addr: s for addr, s in zip(addrs, sockets) } socket_to_addr = { s: addr for addr, s in zip(addrs, sockets) } send_hashes = {} recv_hashes = {} ep = select.epoll() for s in sockets: ep.register(s, select.EPOLLRDNORM) n = 50000 nr_send = 0 nr_recv = 0 while nr_send < n: # Send as much as we can without blocking print("sending...", nr_send, nr_recv) while nr_send < n: send_data = hashlib.sha256( f'packet {nr_send}'.encode('utf-8')).hexdigest().encode('utf-8') # pseudo-random send/receive pattern sender = sockets[nr_send % 2] receiver = sockets[1 - (nr_send % 3) % 2] try: sender.sendto(send_data, socket_to_addr[receiver]) send_hashes.setdefault((sender.fileno(), receiver.fileno()), hashlib.sha256()).update(f'<{send_data}>'.encode('utf-8')) nr_send = nr_send + 1 except BlockingIOError as e: break except OSError as e: if e.errno in [errno.ENOBUFS, errno.ECONNRESET, errno.EPIPE]: break raise # Receive as much as we can without blocking print("receiving...", nr_send, nr_recv) while nr_recv < nr_send: for fileno, eventmask in ep.poll(): receiver = fileno_to_socket[fileno] if eventmask & select.EPOLLRDNORM: while True: try: recv_data, address = receiver.recvfrom(1024) sender = addr_to_socket[address] recv_hashes.setdefault((sender.fileno(), receiver.fileno()), hashlib.sha256()).update( f'<{recv_data}>'.encode('utf-8')) nr_recv = nr_recv + 1 except BlockingIOError as e: break # exercise net/rds/tcp.c:rds_tcp_sysctl_reset() for net in [net0, net1]: ip(f"netns exec {net} /usr/sbin/sysctl net.rds.tcp.rds_tcp_rcvbuf=10000") ip(f"netns exec {net} /usr/sbin/sysctl net.rds.tcp.rds_tcp_sndbuf=10000") print("done", nr_send, nr_recv) # the Python socket module doesn't know these RDS_INFO_FIRST = 10000 RDS_INFO_LAST = 10017 nr_success = 0 nr_error = 0 for s in sockets: for optname in range(RDS_INFO_FIRST, RDS_INFO_LAST + 1): # Sigh, the Python socket module doesn't allow us to pass # buffer lengths greater than 1024 for some reason. RDS # wants multiple pages. try: s.getsockopt(socket.SOL_RDS, optname, 1024) nr_success = nr_success + 1 except OSError as e: nr_error = nr_error + 1 if e.errno == errno.ENOSPC: # ignore pass print(f"getsockopt(): {nr_success}/{nr_error}") print("Stopping network packet captures") subprocess.check_call(['killall', '-q', 'tcpdump']) # We're done sending and receiving stuff, now let's check if what # we received is what we sent. for (sender, receiver), send_hash in send_hashes.items(): recv_hash = recv_hashes.get((sender, receiver)) if recv_hash is None: print("FAIL: No data received") sys.exit(1) if send_hash.hexdigest() != recv_hash.hexdigest(): print("FAIL: Send/recv mismatch") print("hash expected:", send_hash.hexdigest()) print("hash received:", recv_hash.hexdigest()) sys.exit(1) print(f"{sender}/{receiver}: ok") print("Success") sys.exit(0)