#!/usr/bin/env python3 import unittest import os from socket import AF_INET, AF_INET6, inet_pton from framework import VppTestCase, VppTestRunner from vpp_neighbor import VppNeighbor, find_nbr from vpp_ip_route import VppIpRoute, VppRoutePath, find_route, \ VppIpTable, DpoProto, FibPathType from vpp_papi import VppEnum import scapy.compat from scapy.packet import Raw from scapy.layers.l2 import Ether, ARP, Dot1Q from scapy.layers.inet import IP, UDP, TCP from scapy.layers.inet6 import IPv6 from scapy.contrib.mpls import MPLS from scapy.layers.inet6 import IPv6 NUM_PKTS = 67 # not exported by scapy, so redefined here arp_opts = {"who-has": 1, "is-at": 2} class ARPTestCase(VppTestCase): """ ARP Test Case """ @classmethod def setUpClass(cls): super(ARPTestCase, cls).setUpClass() @classmethod def tearDownClass(cls): super(ARPTestCase, cls).tearDownClass() def setUp(self): super(ARPTestCase, self).setUp() # create 3 pg interfaces self.create_pg_interfaces(range(4)) # pg0 configured with ip4 and 6 addresses used for input # pg1 configured with ip4 and 6 addresses used for output # pg2 is unnumbered to pg0 for i in self.pg_interfaces: i.admin_up() self.pg0.config_ip4() self.pg0.config_ip6() self.pg0.resolve_arp() self.pg1.config_ip4() self.pg1.config_ip6() # pg3 in a different VRF self.tbl = VppIpTable(self, 1) self.tbl.add_vpp_config() self.pg3.set_table_ip4(1) self.pg3.config_ip4() def tearDown(self): self.pg0.unconfig_ip4() self.pg0.unconfig_ip6() self.pg1.unconfig_ip4() self.pg1.unconfig_ip6() self.pg3.unconfig_ip4() self.pg3.set_table_ip4(0) for i in self.pg_interfaces: i.admin_down() super(ARPTestCase, self).tearDown() def verify_arp_req(self, rx, smac, sip, dip): ether = rx[Ether] self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff") self.assertEqual(ether.src, smac) arp = rx[ARP] self.assertEqual(arp.hwtype, 1) self.assertEqual(arp.ptype, 0x800) self.assertEqual(arp.hwlen, 6) self.assertEqual(arp.plen, 4) self.assertEqual(arp.op, arp_opts["who-has"]) self.assertEqual(arp.hwsrc, smac) self.assertEqual(arp.hwdst, "00:00:00:00:00:00") self.assertEqual(arp.psrc, sip) self.assertEqual(arp.pdst, dip) def verify_arp_resp(self, rx, smac, dmac, sip, dip): ether = rx[Ether] self.assertEqual(ether.dst, dmac) self.assertEqual(ether.src, smac) arp = rx[ARP] self.assertEqual(arp.hwtype, 1) self.assertEqual(arp.ptype, 0x800) self.assertEqual(arp.hwlen, 6) self.assertEqual(arp.plen, 4) self.assertEqual(arp.op, arp_opts["is-at"]) self.assertEqual(arp.hwsrc, smac) self.assertEqual(arp.hwdst, dmac) self.assertEqual(arp.psrc, sip) self.assertEqual(arp.pdst, dip) def verify_arp_vrrp_resp(self, rx, smac, dmac, sip, dip): ether = rx[Ether] self.assertEqual(ether.dst, dmac) self.assertEqual(ether.src, smac) arp = rx[ARP] self.assertEqual(arp.hwtype, 1) self.assertEqual(arp.ptype, 0x800) self.assertEqual(arp.hwlen, 6) self.assertEqual(arp.plen, 4) self.assertEqual(arp.op, arp_opts["is-at"]) self.assertNotEqual(arp.hwsrc, smac) self.assertTrue("00:00:5e:00:01" in arp.hwsrc or "00:00:5E:00:01" in arp.hwsrc) self.assertEqual(arp.hwdst, dmac) self.assertEqual(arp.psrc, sip) self.assertEqual(arp.pdst, dip) def verify_ip(self, rx, smac, dmac, sip, dip): ether = rx[Ether] self.assertEqual(ether.dst, dmac) self.assertEqual(ether.src, smac) ip = rx[IP] self.assertEqual(ip.src, sip) self.assertEqual(ip.dst, dip) def verify_ip_o_mpls(self, rx, smac, dmac, label, sip, dip): ether = rx[Ether] self.assertEqual(ether.dst, dmac) self.assertEqual(ether.src, smac) mpls = rx[MPLS] self.assertTrue(mpls.label, label) ip = rx[IP] self.assertEqual(ip.src, sip) self.assertEqual(ip.dst, dip) def test_arp(self): """ ARP """ # # Generate some hosts on the LAN # self.pg1.generate_remote_hosts(11) # # watch for: # - all neighbour events # - all neighbor events on pg1 # - neighbor events for host[1] on pg1 # self.vapi.want_ip_neighbor_events(enable=1, pid=os.getpid()) self.vapi.want_ip_neighbor_events(enable=1, pid=os.getpid(), sw_if_index=self.pg1.sw_if_index) self.vapi.want_ip_neighbor_events(enable=1, pid=os.getpid(), sw_if_index=self.pg1.sw_if_index, ip=self.pg1.remote_hosts[1].ip4) self.logger.info(self.vapi.cli("sh ip neighbor-watcher")) # # Send IP traffic to one of these unresolved hosts. # expect the generation of an ARP request # p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1._remote_hosts[1].ip4) / UDP(sport=1234, dport=1234) / Raw()) self.pg0.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_arp_req(rx[0], self.pg1.local_mac, self.pg1.local_ip4, self.pg1._remote_hosts[1].ip4) # # And a dynamic ARP entry for host 1 # dyn_arp = VppNeighbor(self, self.pg1.sw_if_index, self.pg1.remote_hosts[1].mac, self.pg1.remote_hosts[1].ip4) dyn_arp.add_vpp_config() self.assertTrue(dyn_arp.query_vpp_config()) # this matches all of the listnerers es = [self.vapi.wait_for_event(1, "ip_neighbor_event") for i in range(3)] for e in es: self.assertEqual(str(e.neighbor.ip_address), self.pg1.remote_hosts[1].ip4) # # now we expect IP traffic forwarded # dyn_p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1._remote_hosts[1].ip4) / UDP(sport=1234, dport=1234) / Raw()) self.pg0.add_stream(dyn_p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_ip(rx[0], self.pg1.local_mac, self.pg1.remote_hosts[1].mac, self.pg0.remote_ip4, self.pg1._remote_hosts[1].ip4) # # And a Static ARP entry for host 2 # static_arp = VppNeighbor(self, self.pg1.sw_if_index, self.pg1.remote_hosts[2].mac, self.pg1.remote_hosts[2].ip4, is_static=1) static_arp.add_vpp_config() es = [self.vapi.wait_for_event(1, "ip_neighbor_event") for i in range(2)] for e in es: self.assertEqual(str(e.neighbor.ip_address), self.pg1.remote_hosts[2].ip4) static_p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) / IP(src=self.pg0.remote_ip4, dst=self.pg1._remote_hosts[2].ip4) / UDP(sport=1234, dport=1234) / Raw()) self.pg0.add_stream(static_p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_ip(rx[0], self.pg1.local_mac, self.pg1.remote_hosts[2].mac, self.pg0.remote_ip4, self.pg1._remote_hosts[2].ip4) # # remove all the listeners # self.vapi.want_ip_neighbor_events(enable=0, pid=os.getpid()) self.vapi.want_ip_neighbor_events(enable=0, pid=os.getpid(), sw_if_index=self.pg1.sw_if_index) self.vapi.want_ip_neighbor_events(enable=0, pid=os.getpid(), sw_if_index=self.pg1.sw_if_index, ip=self.pg1.remote_hosts[1].ip4) # # flap the link. dynamic ARPs get flush, statics don't # self.pg1.admin_down() self.pg1.admin_up() self.pg0.add_stream(static_p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_ip(rx[0], self.pg1.local_mac, self.pg1.remote_hosts[2].mac, self.pg0.remote_ip4, self.pg1._remote_hosts[2].ip4) self.pg0.add_stream(dyn_p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_arp_req(rx[0], self.pg1.local_mac, self.pg1.local_ip4, self.pg1._remote_hosts[1].ip4) self.assertFalse(dyn_arp.query_vpp_config()) self.assertTrue(static_arp.query_vpp_config()) # # Send an ARP request from one of the so-far unlearned remote hosts # p = (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1._remote_hosts[3].mac) / ARP(op="who-has", hwsrc=self.pg1._remote_hosts[3].mac, pdst=self.pg1.local_ip4, psrc=self.pg1._remote_hosts[3].ip4)) self.pg1.add_stream(p) self.pg_enable_capture(self.pg_interfaces) self.pg_start() rx = self.pg1.get_capture(1) self.verify_arp_resp(rx[0], self.pg1.local_mac, self.pg1._remote_hosts[3].mac, self.pg1.local_ip4, self.pg1._remote_hosts[3].ip4) # # VPP should have learned the mapping for the remote host # self.assertTrue(find_nbr(self, self.pg1.sw_if_index, self.pg1._remote_hosts[3].ip4)) # # Fire in an ARP request before the interface becomes IP enabled # self.pg2.generate_remote_hosts(4) p = (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_mac) / ARP(op="who-has", hwsrc=self.pg2.remote_mac, pdst=self.pg1.local_ip4, psrc=self.pg2.remote_hosts[3].ip4)) pt = (Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_mac) / Dot1Q(vlan=0) / ARP(op="who-has", hwsrc=self.pg2.remote_mac, pdst=self.pg1.local_ip4, psrc=self.pg2.remote_hosts[3].ip4)) self.send_and_assert_no_replies(self.pg2, p, "interface not IP enabled") # # Make pg2 un-numbered to pg1 # self.pg2.set_unnumbered(self.pg1.sw_if_index) # # test the unnumbered dump both by all interfaces and just the enabled # one # unnum = self.vapi.ip_unnumbered_dump() self.assertTrue(len(unnum)) self.assertEqual(unnum[0]
# Copyright (c) 2022 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
/.tox
/archive_dir
/download_dir
/env
# "generated" is a place for autogenerated data, not to pollute git.
/generated
outputs
output.xml
log.html
report.html
*.log
*.pyc
*~
.idea
.vscode
*.raw.json
*.info.json