diff options
Diffstat (limited to 'test/vpp_interface.py')
| -rw-r--r-- | test/vpp_interface.py | 240 |
1 files changed, 240 insertions, 0 deletions
diff --git a/test/vpp_interface.py b/test/vpp_interface.py new file mode 100644 index 00000000000..c596ab5abf7 --- /dev/null +++ b/test/vpp_interface.py @@ -0,0 +1,240 @@ +from abc import abstractmethod, ABCMeta +import socket +from logging import info, error +from scapy.layers.l2 import Ether, ARP + +from scapy.layers.inet6 import IPv6, ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptSrcLLAddr, ICMPv6NDOptDstLLAddr + + +class VppInterface(object): + """ + Generic VPP interface + """ + __metaclass__ = ABCMeta + + @property + def sw_if_index(self): + """Interface index assigned by VPP""" + return self._sw_if_index + + @property + def remote_mac(self): + """MAC-address of the remote interface "connected" to this interface""" + return self._remote_mac + + @property + def local_mac(self): + """MAC-address of the VPP interface""" + return self._local_mac + + @property + def local_ip4(self): + """Local IPv4 address on VPP interface (string)""" + return self._local_ip4 + + @property + def local_ip4n(self): + """Local IPv4 address - raw, suitable as API parameter""" + return self._local_ip4n + + @property + def remote_ip4(self): + """IPv4 address of remote peer "connected" to this interface""" + return self._remote_ip4 + + @property + def remote_ip4n(self): + """IPv4 address of remote peer - raw, suitable as API parameter""" + return self._remote_ip4n + + @property + def local_ip6(self): + """Local IPv6 address on VPP interface (string)""" + return self._local_ip6 + + @property + def local_ip6n(self): + """Local IPv6 address - raw, suitable as API parameter""" + return self._local_ip6n + + @property + def remote_ip6(self): + """IPv6 address of remote peer "connected" to this interface""" + return self._remote_ip6 + + @property + def remote_ip6n(self): + """IPv6 address of remote peer - raw, suitable as API parameter""" + return self._remote_ip6n + + @property + def name(self): + """Name of the interface""" + return self._name + + @property + def dump(self): + """Raw result of sw_interface_dump for this interface""" + return self._dump + + @property + def test(self): + """Test case creating this interface""" + return self._test + + def post_init_setup(self): + """Additional setup run after creating an interface object""" + self._remote_mac = "02:00:00:00:ff:%02x" % self.sw_if_index + + self._local_ip4 = "172.16.%u.1" % self.sw_if_index + self._local_ip4n = socket.inet_pton(socket.AF_INET, self.local_ip4) + self._remote_ip4 = "172.16.%u.2" % self.sw_if_index + self._remote_ip4n = socket.inet_pton(socket.AF_INET, self.remote_ip4) + + self._local_ip6 = "fd01:%u::1" % self.sw_if_index + self._local_ip6n = socket.inet_pton(socket.AF_INET6, self.local_ip6) + self._remote_ip6 = "fd01:%u::2" % self.sw_if_index + self._remote_ip6n = socket.inet_pton(socket.AF_INET6, self.remote_ip6) + + r = self.test.vapi.sw_interface_dump() + for intf in r: + if intf.sw_if_index == self.sw_if_index: + self._name = intf.interface_name.split(b'\0', 1)[0] + self._local_mac = ':'.join(intf.l2_address.encode('hex')[i:i + 2] + for i in range(0, 12, 2)) + self._dump = intf + break + else: + raise Exception( + "Could not find interface with sw_if_index %d " + "in interface dump %s" % + (self.sw_if_index, repr(r))) + + @abstractmethod + def __init__(self, test, index): + self._test = test + self.post_init_setup() + info("New %s, MAC=%s, remote_ip4=%s, local_ip4=%s" % + (self.__name__, self.remote_mac, self.remote_ip4, self.local_ip4)) + + def config_ip4(self): + """Configure IPv4 address on the VPP interface""" + addr = self.local_ip4n + addr_len = 24 + self.test.vapi.sw_interface_add_del_address( + self.sw_if_index, addr, addr_len) + + def config_ip6(self): + """Configure IPv6 address on the VPP interface""" + addr = self._local_ip6n + addr_len = 64 + self.test.vapi.sw_interface_add_del_address( + self.sw_if_index, addr, addr_len, is_ipv6=1) + + def disable_ipv6_ra(self): + """Configure IPv6 RA suppress on the VPP interface""" + self.test.vapi.sw_interface_ra_suppress(self.sw_if_index) + + def create_arp_req(self): + """Create ARP request applicable for this interface""" + return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) / + ARP(op=ARP.who_has, pdst=self.local_ip4, + psrc=self.remote_ip4, hwsrc=self.remote_mac)) + + def create_ndp_req(self): + return (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.remote_mac) / + IPv6(src=self.remote_ip6, dst=self.local_ip6) / + ICMPv6ND_NS(tgt=self.local_ip6) / + ICMPv6NDOptSrcLLAddr(lladdr=self.remote_mac)) + + def resolve_arp(self, pg_interface=None): + """Resolve ARP using provided packet-generator interface + + :param pg_interface: interface used to resolve, if None then this + interface is used + + """ + if pg_interface is None: + pg_interface = self + info("Sending ARP request for %s on port %s" % + (self.local_ip4, pg_interface.name)) + arp_req = self.create_arp_req() + pg_interface.add_stream(arp_req) + pg_interface.enable_capture() + self.test.pg_start() + info(self.test.vapi.cli("show trace")) + arp_reply = pg_interface.get_capture() + if arp_reply is None or len(arp_reply) == 0: + info("No ARP received on port %s" % pg_interface.name) + return + arp_reply = arp_reply[0] + # Make Dot1AD packet content recognizable to scapy + if arp_reply.type == 0x88a8: + arp_reply.type = 0x8100 + arp_reply = Ether(str(arp_reply)) + try: + if arp_reply[ARP].op == ARP.is_at: + info("VPP %s MAC address is %s " % + (self.name, arp_reply[ARP].hwsrc)) + self._local_mac = arp_reply[ARP].hwsrc + else: + info("No ARP received on port %s" % pg_interface.name) + except: + error("Unexpected response to ARP request:") + error(arp_reply.show()) + raise + + def resolve_ndp(self, pg_interface=None): + """Resolve NDP using provided packet-generator interface + + :param pg_interface: interface used to resolve, if None then this + interface is used + + """ + if pg_interface is None: + pg_interface = self + info("Sending NDP request for %s on port %s" % + (self.local_ip6, pg_interface.name)) + ndp_req = self.create_ndp_req() + pg_interface.add_stream(ndp_req) + pg_interface.enable_capture() + self.test.pg_start() + info(self.test.vapi.cli("show trace")) + ndp_reply = pg_interface.get_capture() + if ndp_reply is None or len(ndp_reply) == 0: + info("No NDP received on port %s" % pg_interface.name) + return + ndp_reply = ndp_reply[0] + # Make Dot1AD packet content recognizable to scapy + if ndp_reply.type == 0x88a8: + ndp_reply.type = 0x8100 + ndp_reply = Ether(str(ndp_reply)) + try: + ndp_na = ndp_reply[ICMPv6ND_NA] + opt = ndp_na[ICMPv6NDOptDstLLAddr] + info("VPP %s MAC address is %s " % + (self.name, opt.lladdr)) + self._local_mac = opt.lladdr + except: + error("Unexpected response to NDP request:") + error(ndp_reply.show()) + raise + + def admin_up(self): + """ Put interface ADMIN-UP """ + self.test.vapi.sw_interface_set_flags(self.sw_if_index, admin_up_down=1) + + def add_sub_if(self, sub_if): + """ + Register a sub-interface with this interface + + :param sub_if: sub-interface + + """ + if not hasattr(self, 'sub_if'): + self.sub_if = sub_if + else: + if isinstance(self.sub_if, list): + self.sub_if.append(sub_if) + else: + self.sub_if = sub_if |