| #! /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) |
