Convert arp_responder.py script to python 3 (#8697)

This allows the use of scapy 2.5.0 and also fixes an issue where ARP packets are not being sent out fast enough/in time for control plane ping packets to fail to respond. Signed-off-by: Saikrishna Arcot <[email protected]>
sonic-net · Jul 25, 2023 · 01a0a5f · 01a0a5f
1 parent 370f788
commit 01a0a5f
Show file tree

Hide file tree

Showing 3 changed files with 61 additions and 201 deletions.
diff --git a/ansible/roles/test/templates/arp_responder.conf.j2 b/ansible/roles/test/templates/arp_responder.conf.j2
@@ -1,5 +1,5 @@
 [program:arp_responder]
-command=/usr/bin/python /opt/arp_responder.py {{ arp_responder_args }}
+command=/root/env-python3/bin/python3 /opt/arp_responder.py {{ arp_responder_args }}
 process_name=arp_responder
 stdout_logfile=/tmp/arp_responder.out.log
 stderr_logfile=/tmp/arp_responder.err.log

diff --git a/tests/scripts/arp_responder.py b/tests/scripts/arp_responder.py
@@ -1,215 +1,85 @@
 import binascii
-import socket
-import struct
-import select
 import json
 import argparse
 import os.path
 from collections import defaultdict
-from fcntl import ioctl
 import logging
-import ptf.packet as scapy
-import scapy.all as scapy2
-scapy2.conf.use_pcap = True        # Do not move this import. use_pcap import needs to be enabled before import pcapdnet
-import scapy.arch.pcapdnet # noqa F401
-
+import scapy.all as scapy
 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
 
 
-NEIGH_SOLICIT_ICMP_MSG_TYPE = 135
-
-
-def hexdump(data):
-    print((" ".join("%02x" % ord(d) for d in data)))
-
-
-def get_if(iff, cmd):
-    s = socket.socket()
-    ifreq = ioctl(s, cmd, struct.pack("16s16x", iff))
-    s.close()
-
-    return ifreq
-
-
-def get_mac(iff):
-    SIOCGIFHWADDR = 0x8927          # Get hardware address
-    return get_if(iff, SIOCGIFHWADDR)[18:24]
-
-
-class Interface(object):
-    ETH_P_ALL = 0x03
-    ETH_P_ARP = 0x806
-    RCV_TIMEOUT = 1000
-    RCV_SIZE = 4096
-
-    def __init__(self, iface):
-        self.iface = iface
-        self.socket = None
-        self.mac_address = get_mac(iface)
-
-    def __del__(self):
-        if self.socket:
-            self.socket.close()
-
-    def bind(self):
-        self.socket = scapy2.conf.L2listen(
-            iface=self.iface, filter='arp || ip6[40] = {}'.format(NEIGH_SOLICIT_ICMP_MSG_TYPE))
-
-    def handler(self):
-        return self.socket
-
-    def recv(self):
-        sniffed = self.socket.recv()
-        pkt = sniffed[0]
-        str_pkt = str(pkt).encode("HEX")
-        binpkt = binascii.unhexlify(str_pkt)
-        return binpkt
-
-    def send(self, data):
-        scapy2.sendp(data, iface=self.iface)
-
-    def mac(self):
-        return self.mac_address
-
-    def name(self):
-        return self.iface
-
-
-class Poller(object):
-    def __init__(self, interfaces, responder):
-        self.responder = responder
-        self.mapping = {}
-        for interface in interfaces:
-            self.mapping[interface.handler()] = interface
-
-    def poll(self):
-        handlers = list(self.mapping.keys())
-        while True:
-            (rdlist, _, _) = select.select(handlers, [], [])
-            for handler in rdlist:
-                self.responder.action(self.mapping[handler])
-
-
 class ARPResponder(object):
-    ARP_PKT_LEN = 64
-    NDP_PKT_LEN = 90
     ARP_OP_REQUEST = 1
+    ip_sets = {}
 
-    def __init__(self, ip_sets):
-        # defines a part of the packet for ARP Reply
-        self.arp_chunk = binascii.unhexlify('08060001080006040002')
-        self.arp_pad = binascii.unhexlify('00' * 18)
-
-        self.ip_sets = ip_sets
-
-        return
-
-    def action(self, interface):
-        data = interface.recv()
-        eth_type = struct.unpack('!H', data[12:14])[0]
-        # Retrieve the correct ethertype if the packet is VLAN tagged
-        if eth_type == 0x8100:  # 802.1Q, VLAN tagged
-            eth_type = struct.unpack('!H', data[16:18])[0]
-
-        if eth_type == 0x0806:  # ARP
-            if len(data) <= self.ARP_PKT_LEN:
-                return self.reply_to_arp(data, interface)
-            else:
-                # Handle the case where data length is greater than ARP packet length
-                pass
-        elif eth_type == 0x86DD:  # IPv6
-            if len(data) <= self.NDP_PKT_LEN:
-                return self.reply_to_ndp(data, interface)
-            else:
-                # Handle the case where data length is greater than NDP packet length
-                pass
+    @staticmethod
+    def action(packet):
+        if "ARP" in packet:  # IPv4
+            return ARPResponder.reply_to_arp(packet)
+        elif "ICMPv6ND_NS" in packet and "ICMPv6NDOptSrcLLAddr" in packet:  # IPv6
+            return ARPResponder.reply_to_ndp(packet)
         else:
             # Handle other Ethernet types
             pass
 
-    def reply_to_arp(self, data, interface):
-        remote_mac, remote_ip, request_ip, op_type, vlan_id = self.extract_arp_info(
-            data)
+    @staticmethod
+    def reply_to_arp(data):
+        remote_mac = data["ARP"].hwsrc
+        remote_ip = data["ARP"].psrc
+        request_ip = data["ARP"].pdst
+        op_type = data["ARP"].op
 
         # Don't send ARP response if the ARP op code is not request
-        if op_type != self.ARP_OP_REQUEST:
+        if op_type != ARPResponder.ARP_OP_REQUEST:
             return
 
-        request_ip_str = socket.inet_ntoa(request_ip)
-        if request_ip_str not in self.ip_sets[interface.name()]:
+        interface = data.sniffed_on
+        if interface not in ARPResponder.ip_sets:
+            return
+        if request_ip not in ARPResponder.ip_sets[interface]:
             return
 
-        if 'vlan' in self.ip_sets[interface.name()]:
-            vlan_list = self.ip_sets[interface.name()]['vlan']
+        if 'vlan' in ARPResponder.ip_sets[interface]:
+            vlan_list = ARPResponder.ip_sets[interface]['vlan']
         else:
             vlan_list = [None]
 
         for vlan_id in vlan_list:
-            arp_reply = self.generate_arp_reply(self.ip_sets[interface.name(
-            )][request_ip_str], remote_mac, request_ip, remote_ip, vlan_id)
-            interface.send(arp_reply)
-
-        return
-
-    def reply_to_ndp(self, data, interface):
-        remote_mac, remote_ip, target_ip = self.extract_ndp_info(data)
-
-        target_ip_str = socket.inet_ntop(socket.AF_INET6, target_ip)
-        if target_ip_str in self.ip_sets[interface.name()]:
-            remote_ip_str = socket.inet_ntop(socket.AF_INET6, remote_ip)
-            neigh_adv_pkt = self.generate_neigh_adv(self.ip_sets[interface.name(
-            )][target_ip_str], remote_mac, target_ip_str, remote_ip_str)
-            interface.send(neigh_adv_pkt)
-
-        return
-
-    def extract_ndp_info(self, data):
-        vlan_offset = 0
-
-        if len(data) == 90:
-            vlan_offset = 4
-
-        remote_mac = data[6:12]
-        remote_ip = data[22 + vlan_offset:38 + vlan_offset]
-        target_ip = data[62 + vlan_offset:78 + vlan_offset]
-
-        return remote_mac, remote_ip, target_ip
-
-    def extract_arp_info(self, data):
-        # remote_mac, remote_ip, request_ip, op_type
-        rem_ip_start = 28
-        req_ip_start = 38
-        op_type_start = 20
-        eth_offset = 0
-        vlan_id = None
-        ether_type = str(data[12:14]).encode("HEX")
-        if (ether_type == '8100'):
-            vlan = str(data[14:16]).encode("HEX")
-            if (vlan != '0000'):
-                eth_offset = 4
-                vlan_id = data[14:16]
-        rem_ip_start = rem_ip_start + eth_offset
-        req_ip_start = req_ip_start + eth_offset
-        op_type_start = op_type_start + eth_offset
-        rem_ip_end = rem_ip_start + 4
-        req_ip_end = req_ip_start + 4
-        op_type_end = op_type_start + 1
+            arp_reply = ARPResponder.generate_arp_reply(ARPResponder.ip_sets[interface][request_ip],
+                                                        remote_mac, request_ip, remote_ip, vlan_id)
+            scapy.sendp(arp_reply, iface=interface)
+
+    @staticmethod
+    def reply_to_ndp(data):
+        remote_mac = data["ICMPv6NDOptSrcLLAddr"].lladdr
+        remote_ip = data["IPv6"].src
+        request_ip = data["ICMPv6ND_NS"].tgt
+
+        interface = data.sniffed_on
+        if interface not in ARPResponder.ip_sets:
+            return
+        if request_ip not in ARPResponder.ip_sets[interface]:
+            return
 
-        return data[6:12], data[rem_ip_start:rem_ip_end], data[req_ip_start:req_ip_end],\
-            (ord(data[op_type_start]) * 256 + ord(data[op_type_end])), vlan_id
+        ndp_reply = ARPResponder.generate_neigh_adv(ARPResponder.ip_sets[interface][request_ip],
+                                                    remote_mac, request_ip, remote_ip)
+        scapy.sendp(ndp_reply)
 
-    def generate_arp_reply(self, local_mac, remote_mac, local_ip, remote_ip, vlan_id):
-        eth_hdr = remote_mac + local_mac
-        if vlan_id is not None:
-            eth_type = binascii.unhexlify('8100')
-            eth_hdr += eth_type + vlan_id
+    @staticmethod
+    def generate_arp_reply(local_mac, remote_mac, local_ip, remote_ip, vlan_id):
+        l2 = scapy.Ether(dst=remote_mac, src=local_mac, type=(0x8100 if vlan_id else 0x0806))
+        l3 = scapy.ARP(op=2, hwsrc=local_mac, psrc=local_ip, hwdst=remote_mac, pdst=remote_ip)
+        if vlan_id:
+            l2 /= scapy.Dot1Q(vlan=vlan_id, type=0x0806)
 
-        return eth_hdr + self.arp_chunk + local_mac + local_ip + remote_mac + remote_ip + self.arp_pad
+        return l2 / l3
 
+    @staticmethod
     def generate_neigh_adv(self, local_mac, remote_mac, target_ip, remote_ip):
-        neigh_adv_pkt = Ether(src=local_mac, dst=remote_mac) / IPv6(src=target_ip, dst=remote_ip) # noqa F821
-        neigh_adv_pkt /= ICMPv6ND_NA(tgt=target_ip, R=0, S=1, O=1)                                # noqa F821
-        neigh_adv_pkt /= ICMPv6NDOptDstLLAddr(lladdr=local_mac)                                   # noqa F821
+        neigh_adv_pkt = scapy.Ether(src=local_mac, dst=remote_mac)
+        neigh_adv_pkt /= scapy.IPv6(src=target_ip, dst=remote_ip)
+        neigh_adv_pkt /= scapy.ICMPv6ND_NA(tgt=target_ip, R=0, S=1, O=1)
+        neigh_adv_pkt /= scapy.ICMPv6NDOptDstLLAddr(lladdr=local_mac)
 
         return neigh_adv_pkt
 
@@ -237,37 +107,27 @@ def main():
 
     # generate ip_sets. every ip address will have it's own uniq mac address
     ip_sets = {}
-    counter = 0
     for iface, ip_dict in list(data.items()):
         vlan = None
+        iface = str(iface)
         if iface.find('@') != -1:
             iface, vlan = iface.split('@')
             vlan_tag = format(int(vlan), 'x')
             vlan_tag = vlan_tag.zfill(4)
-        if str(iface) not in ip_sets:
-            ip_sets[str(iface)] = defaultdict(list)
+        if iface not in ip_sets:
+            ip_sets[iface] = defaultdict(list)
         if args.extended:
             for ip, mac in list(ip_dict.items()):
-                ip_sets[str(iface)][str(ip)] = binascii.unhexlify(str(mac))
-                counter += 1
+                ip_sets[iface][str(ip)] = binascii.unhexlify(str(mac))
         else:
             for ip in ip_dict:
-                ip_sets[str(iface)][str(ip)] = get_mac(str(iface))
+                ip_sets[iface][str(ip)] = scapy.get_if_hwaddr(iface)
         if vlan is not None:
-            ip_sets[str(iface)]['vlan'].append(binascii.unhexlify(vlan_tag))
-
-    ifaces = []
-    for iface_name in list(ip_sets.keys()):
-        iface = Interface(iface_name)
-        iface.bind()
-        ifaces.append(iface)
-
-    resp = ARPResponder(ip_sets)
+            ip_sets[iface]['vlan'].append(binascii.unhexlify(vlan_tag))
 
-    p = Poller(ifaces, resp)
-    p.poll()
+    ARPResponder.ip_sets = ip_sets
 
-    return
+    scapy.sniff(prn=ARPResponder.action, filter="arp or icmp6", iface=list(ip_sets.keys()), store=False)
 
 
 if __name__ == '__main__':

diff --git a/tests/templates/arp_responder.conf.j2 b/tests/templates/arp_responder.conf.j2
@@ -1,5 +1,5 @@
 [program:arp_responder]
-command=/usr/bin/python /opt/arp_responder.py {{ arp_responder_args }}
+command=/root/env-python3/bin/python3 /opt/arp_responder.py {{ arp_responder_args }}
 process_name=arp_responder
 stdout_logfile=/tmp/arp_responder.out.log
 stderr_logfile=/tmp/arp_responder.err.log