from vpp_ip_route import VppIpRoute, VppRoutePath, find_route
from scapy.packet import Raw
-from scapy.layers.l2 import Ether, ARP
+from scapy.layers.l2 import Ether, ARP, Dot1Q
from scapy.layers.inet import IP, UDP
from scapy.contrib.mpls import MPLS
#
# Generate some hosts on the LAN
#
- self.pg1.generate_remote_hosts(9)
+ self.pg1.generate_remote_hosts(10)
#
# Send IP traffic to one of these unresolved hosts.
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")
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
+ rx = self.pg2.get_capture(1)
+ self.verify_arp_resp(rx[0],
+ self.pg2.local_mac,
+ self.pg2.remote_mac,
+ self.pg1.local_ip4,
+ self.pg2.remote_hosts[3].ip4)
+
+ self.pg2.add_stream(pt)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
rx = self.pg2.get_capture(1)
self.verify_arp_resp(rx[0],
self.pg2.local_mac,
self.pg1.local_ip4,
self.pg1.remote_hosts[8].ip4)
+ #
+ # Send an ARP request from one of the so-far unlearned remote hosts
+ # with a VLAN0 tag
+ #
+ p = (Ether(dst="ff:ff:ff:ff:ff:ff",
+ src=self.pg1._remote_hosts[9].mac) /
+ Dot1Q(vlan=0) /
+ ARP(op="who-has",
+ hwsrc=self.pg1._remote_hosts[9].mac,
+ pdst=self.pg1.local_ip4,
+ psrc=self.pg1._remote_hosts[9].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[9].mac,
+ self.pg1.local_ip4,
+ self.pg1._remote_hosts[9].ip4)
+
#
# ERROR Cases
# 1 - don't respond to ARP request for address not within the
hwsrc=self.pg0.remote_mac,
pdst="10.10.10.3",
psrc=self.pg0.remote_ip4))
+ arp_req_pg0_tagged = (Ether(src=self.pg0.remote_mac,
+ dst="ff:ff:ff:ff:ff:ff") /
+ Dot1Q(vlan=0) /
+ ARP(op="who-has",
+ hwsrc=self.pg0.remote_mac,
+ pdst="10.10.10.3",
+ psrc=self.pg0.remote_ip4))
arp_req_pg1 = (Ether(src=self.pg1.remote_mac,
dst="ff:ff:ff:ff:ff:ff") /
ARP(op="who-has",
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
+ rx = self.pg0.get_capture(1)
+ self.verify_arp_resp(rx[0],
+ self.pg0.local_mac,
+ self.pg0.remote_mac,
+ "10.10.10.3",
+ self.pg0.remote_ip4)
+
+ self.pg0.add_stream(arp_req_pg0_tagged)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
rx = self.pg0.get_capture(1)
self.verify_arp_resp(rx[0],
self.pg0.local_mac,
self.pg1.admin_down()
self.pg1.admin_up()
+ def test_arp_duplicates(self):
+ """ ARP Duplicates"""
+
+ #
+ # Generate some hosts on the LAN
+ #
+ self.pg1.generate_remote_hosts(3)
+
+ #
+ # Add host 1 on pg1 and pg2
+ #
+ arp_pg1 = VppNeighbor(self,
+ self.pg1.sw_if_index,
+ self.pg1.remote_hosts[1].mac,
+ self.pg1.remote_hosts[1].ip4)
+ arp_pg1.add_vpp_config()
+ arp_pg2 = VppNeighbor(self,
+ self.pg2.sw_if_index,
+ self.pg2.remote_mac,
+ self.pg1.remote_hosts[1].ip4)
+ arp_pg2.add_vpp_config()
+
+ #
+ # IP packet destined for pg1 remote host arrives on pg1 again.
+ #
+ 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()
+
+ rx1 = self.pg1.get_capture(1)
+
+ self.verify_ip(rx1[0],
+ self.pg1.local_mac,
+ self.pg1.remote_hosts[1].mac,
+ self.pg0.remote_ip4,
+ self.pg1.remote_hosts[1].ip4)
+
+ #
+ # remove the duplicate on pg1
+ # packet stream shoud generate ARPs out of pg1
+ #
+ arp_pg1.remove_vpp_config()
+
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx1 = self.pg1.get_capture(1)
+
+ self.verify_arp_req(rx1[0],
+ self.pg1.local_mac,
+ self.pg1.local_ip4,
+ self.pg1.remote_hosts[1].ip4)
+
+ #
+ # Add it back
+ #
+ arp_pg1.add_vpp_config()
+
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ rx1 = self.pg1.get_capture(1)
+
+ self.verify_ip(rx1[0],
+ self.pg1.local_mac,
+ self.pg1.remote_hosts[1].mac,
+ self.pg0.remote_ip4,
+ self.pg1.remote_hosts[1].ip4)
+
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)