ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
vpp/test/test_neighbor.py
Dave Wallace 8800f732f8 tests: refactor asf framework code 
- Make framework.py classes a subset of asfframework.py classes
- Remove all packet related code from asfframework.py
- Add test class and test case set up debug output to log
- Repatriate packet tests from asf to test directory
- Remove non-packet related code from framework.py and
  inherit them from asfframework.py classes
- Clean up unused import variables
- Re-enable BFD tests on Ubuntu 22.04 and fix
  intermittent test failures in echo_looped_back
  testcases (where # control packets verified but
  not guaranteed to be received during test)
- Re-enable Wireguard tests on Ubuntu 22.04 and fix
  intermittent test failures in handshake ratelimiting
  testcases and event testcase
- Run Wiregard testcase suites solo
- Improve debug output in log.txt
- Increase VCL/LDP post sleep timeout to allow iperf server
  to finish cleanly.
- Fix pcap history files to be sorted by suite and testcase
  and ensure order/timestamp is correct based on creation
  in the testcase.
- Decode pcap files for each suite and testcase for all
  errors or if configured via comandline option / env var
- Improve vpp corefile detection to allow complete corefile
  generation
- Disable vm vpp interfaces testcases on debian11
- Clean up failed unittest dir when retrying failed testcases
  and unify testname directory and failed linknames into
  framwork functions

Type: test

Change-Id: I0764f79ea5bb639d278bf635ed2408d4d5220e1e
Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
2023-11-03 05:06:43 +00:00

2826 lines
85 KiB
Python
Raw Blame History

#!/usr/bin/env python3
import unittest
import os
from framework import VppTestCase
from asfframework import VppTestRunner, tag_fixme_vpp_workers, tag_fixme_ubuntu2204
from vpp_neighbor import VppNeighbor, find_nbr
from vpp_ip_route import (
VppIpRoute,
VppRoutePath,
find_route,
VppIpTable,
DpoProto,
FibPathType,
VppIpInterfaceAddress,
)
from vpp_papi import VppEnum, MACAddress
from vpp_ip import VppIpPuntRedirect
from vpp_sub_interface import VppDot1ADSubint
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, etype=0x0806):
ether = rx[Ether]
self.assertEqual(ether.dst, "ff:ff:ff:ff:ff:ff")
self.assertEqual(ether.src, smac)
self.assertEqual(ether.type, etype)
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)
self.assertEqual(ether.type, 0x0806)
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)
self.assertEqual(ether.type, 0x0800)
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)
self.assertEqual(ether.type, 0x8847)
mpls = rx[MPLS]
self.assertTrue(mpls.label, label)
ip = rx[IP]
self.assertEqual(ip.src, sip)
self.assertEqual(ip.dst, dip)
def get_arp_rx_requests(self, itf):
"""Get ARP RX request stats for and interface"""
return self.statistics["/net/arp/rx/requests"][:, itf.sw_if_index].sum()
def get_arp_tx_requests(self, itf):
"""Get ARP TX request stats for and interface"""
return self.statistics["/net/arp/tx/requests"][:, itf.sw_if_index].sum()
def get_arp_rx_replies(self, itf):
"""Get ARP RX replies stats for and interface"""
return self.statistics["/net/arp/rx/replies"][:, itf.sw_if_index].sum()
def get_arp_tx_replies(self, itf):
"""Get ARP TX replies stats for and interface"""
return self.statistics["/net/arp/tx/replies"][:, itf.sw_if_index].sum()
def get_arp_rx_garp(self, itf):
"""Get ARP RX grat stats for and interface"""
return self.statistics["/net/arp/rx/gratuitous"][:, itf.sw_if_index].sum()
def get_arp_tx_garp(self, itf):
"""Get ARP RX grat stats for and interface"""
return self.statistics["/net/arp/tx/gratuitous"][:, itf.sw_if_index].sum()
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
)
self.logger.info(self.vapi.cli("sh ip neighbor-stats"))
self.logger.info(self.vapi.cli("sh ip neighbor-stats pg1"))
self.assert_equal(self.get_arp_tx_requests(self.pg1), 1)
#
# 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())
self.logger.info(self.vapi.cli("show ip neighbor-watcher"))
# 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.assert_equal(self.get_arp_tx_requests(self.pg1), 2)
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,
)
self.logger.info(self.vapi.cli("sh ip neighbor-stats pg1"))
self.assert_equal(self.get_arp_rx_requests(self.pg1), 1)
self.assert_equal(self.get_arp_tx_replies(self.pg1), 1)
#
# 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].ip_sw_if_index, self.pg1.sw_if_index)
self.assertEqual(unnum[0].sw_if_index, self.pg2.sw_if_index)
unnum = self.vapi.ip_unnumbered_dump(self.pg2.sw_if_index)
self.assertTrue(len(unnum))
self.assertEqual(unnum[0].ip_sw_if_index, self.pg1.sw_if_index)
self.assertEqual(unnum[0].sw_if_index, self.pg2.sw_if_index)
#
# We should respond to ARP requests for the unnumbered to address
# once an attached route to the source is known
#
self.send_and_assert_no_replies(
self.pg2, p, "ARP req for unnumbered address - no source"
)
attached_host = VppIpRoute(
self,
self.pg2.remote_hosts[3].ip4,
32,
[VppRoutePath("0.0.0.0", self.pg2.sw_if_index)],
)
attached_host.add_vpp_config()
self.pg2.add_stream(p)
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.pg2.remote_mac,
self.pg1.local_ip4,
self.pg2.remote_hosts[3].ip4,
)
#
# A neighbor entry that has no associated FIB-entry
#
arp_no_fib = VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[4].mac,
self.pg1.remote_hosts[4].ip4,
is_no_fib_entry=1,
)
arp_no_fib.add_vpp_config()
#
# check we have the neighbor, but no route
#
self.assertTrue(
find_nbr(self, self.pg1.sw_if_index, self.pg1._remote_hosts[4].ip4)
)
self.assertFalse(find_route(self, self.pg1._remote_hosts[4].ip4, 32))
#
# pg2 is unnumbered to pg1, so we can form adjacencies out of pg2
# from within pg1's subnet
#
arp_unnum = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg1.remote_hosts[5].mac,
self.pg1.remote_hosts[5].ip4,
)
arp_unnum.add_vpp_config()
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=self.pg1._remote_hosts[5].ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg2.get_capture(1)
self.verify_ip(
rx[0],
self.pg2.local_mac,
self.pg1.remote_hosts[5].mac,
self.pg0.remote_ip4,
self.pg1._remote_hosts[5].ip4,
)
#
# ARP requests from hosts in pg1's subnet sent on pg2 are replied to
# with the unnumbered interface's address as the source
#
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.pg1.remote_hosts[6].ip4,
)
self.pg2.add_stream(p)
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.pg1.remote_hosts[6].ip4,
)
#
# An attached host route out of pg2 for an undiscovered hosts generates
# an ARP request with the unnumbered address as the source
#
att_unnum = VppIpRoute(
self,
self.pg1.remote_hosts[7].ip4,
32,
[VppRoutePath("0.0.0.0", self.pg2.sw_if_index)],
)
att_unnum.add_vpp_config()
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=self.pg1._remote_hosts[7].ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg2.get_capture(1)
self.verify_arp_req(
rx[0], self.pg2.local_mac, self.pg1.local_ip4, self.pg1._remote_hosts[7].ip4
)
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.pg1.remote_hosts[7].ip4,
)
self.pg2.add_stream(p)
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.pg1.remote_hosts[7].ip4,
)
#
# An attached host route as yet unresolved out of pg2 for an
# undiscovered host, an ARP requests begets a response.
#
att_unnum1 = VppIpRoute(
self,
self.pg1.remote_hosts[8].ip4,
32,
[VppRoutePath("0.0.0.0", self.pg2.sw_if_index)],
)
att_unnum1.add_vpp_config()
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.pg1.remote_hosts[8].ip4,
)
self.pg2.add_stream(p)
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.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,
)
#
# Add a hierarchy of routes for a host in the sub-net.
# Should still get an ARP resp since the cover is attached
#
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg1.remote_mac,
pdst=self.pg1.local_ip4,
psrc=self.pg1.remote_hosts[10].ip4,
)
r1 = VppIpRoute(
self,
self.pg1.remote_hosts[10].ip4,
30,
[VppRoutePath(self.pg1.remote_hosts[10].ip4, self.pg1.sw_if_index)],
)
r1.add_vpp_config()
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_mac,
self.pg1.local_ip4,
self.pg1.remote_hosts[10].ip4,
)
r2 = VppIpRoute(
self,
self.pg1.remote_hosts[10].ip4,
32,
[VppRoutePath(self.pg1.remote_hosts[10].ip4, self.pg1.sw_if_index)],
)
r2.add_vpp_config()
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_mac,
self.pg1.local_ip4,
self.pg1.remote_hosts[10].ip4,
)
#
# add an ARP entry that's not on the sub-net and so whose
# adj-fib fails the refinement check. then send an ARP request
# from that source
#
a1 = VppNeighbor(
self, self.pg0.sw_if_index, self.pg0.remote_mac, "100.100.100.50"
)
a1.add_vpp_config()
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
psrc="100.100.100.50",
pdst=self.pg0.remote_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req for from failed adj-fib")
#
# ERROR Cases
# 1 - don't respond to ARP request for address not within the
# interface's sub-net
# 1b - nor within the unnumbered subnet
# 1c - nor within the subnet of a different interface
#
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
pdst="10.10.10.3",
psrc=self.pg0.remote_ip4,
)
self.send_and_assert_no_replies(
self.pg0, p, "ARP req for non-local destination"
)
self.assertFalse(find_nbr(self, self.pg0.sw_if_index, "10.10.10.3"))
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg2.remote_mac,
pdst="10.10.10.3",
psrc=self.pg1.remote_hosts[7].ip4,
)
self.send_and_assert_no_replies(
self.pg0, p, "ARP req for non-local destination - unnum"
)
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
pdst=self.pg1.local_ip4,
psrc=self.pg1.remote_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req diff sub-net")
self.assertFalse(find_nbr(self, self.pg0.sw_if_index, self.pg1.remote_ip4))
#
# 2 - don't respond to ARP request from an address not within the
# interface's sub-net
# 2b - to a proxied address
# 2c - not within a different interface's sub-net
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
psrc="10.10.10.3",
pdst=self.pg0.local_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req for non-local source")
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg2.remote_mac,
psrc="10.10.10.3",
pdst=self.pg0.local_ip4,
)
self.send_and_assert_no_replies(
self.pg0, p, "ARP req for non-local source - unnum"
)
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
psrc=self.pg1.remote_ip4,
pdst=self.pg0.local_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req for non-local source 2c")
#
# 3 - don't respond to ARP request from an address that belongs to
# the router
#
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
psrc=self.pg0.local_ip4,
pdst=self.pg0.local_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req for non-local source")
#
# 4 - don't respond to ARP requests that has mac source different
# from ARP request HW source
#
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc="00:00:00:DE:AD:BE",
psrc=self.pg0.remote_ip4,
pdst=self.pg0.local_ip4,
)
self.send_and_assert_no_replies(self.pg0, p, "ARP req for non-local source")
#
# 5 - don't respond to ARP requests for address within the
# interface's sub-net but not the interface's address
#
self.pg0.generate_remote_hosts(2)
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwsrc=self.pg0.remote_mac,
psrc=self.pg0.remote_hosts[0].ip4,
pdst=self.pg0.remote_hosts[1].ip4,
)
self.send_and_assert_no_replies(
self.pg0, p, "ARP req for non-local destination"
)
#
# cleanup
#
static_arp.remove_vpp_config()
self.pg2.unset_unnumbered(self.pg1.sw_if_index)
# need this to flush the adj-fibs
self.pg2.unset_unnumbered(self.pg1.sw_if_index)
self.pg2.admin_down()
self.pg1.admin_down()
def test_arp_after_mac_change(self):
"""ARP (after MAC address change)"""
#
# Prepare a subinterface
#
subif0 = VppDot1ADSubint(self, self.pg1, 0, 300, 400)
subif0.admin_up()
subif0.config_ip4()
#
# Send a packet to cause ARP generation for the parent interface's remote host
#
p1 = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg0.add_stream(p1)
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_ip4
)
#
# Send a packet to cause ARP generation for the subinterface's remote host
#
p2 = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=subif0.remote_ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg0.add_stream(p2)
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,
subif0.local_ip4,
subif0.remote_ip4,
subif0.DOT1AD_TYPE,
)
#
# Change MAC address of the parent interface
#
pg1_mac_saved = self.pg1.local_mac
self.pg1.set_mac(MACAddress("00:00:00:11:22:33"))
#
# Send a packet to cause ARP generation for the parent interface's remote host
# - expect new MAC address is used as the source
#
self.pg0.add_stream(p1)
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_ip4
)
#
# Send a packet to cause ARP generation for the subinterface's remote host
# - expect new MAC address is used as the source
#
self.pg0.add_stream(p2)
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,
subif0.local_ip4,
subif0.remote_ip4,
subif0.DOT1AD_TYPE,
)
#
# Cleanup
#
subif0.remove_vpp_config()
self.pg1.set_mac(MACAddress(pg1_mac_saved))
def test_proxy_mirror_arp(self):
"""Interface Mirror Proxy ARP"""
#
# When VPP has an interface whose address is also applied to a TAP
# interface on the host, then VPP's TAP interface will be unnumbered
# to the 'real' interface and do proxy ARP from the host.
# the curious aspect of this setup is that ARP requests from the host
# will come from the VPP's own address.
#
self.pg0.generate_remote_hosts(2)
arp_req_from_me = Ether(src=self.pg2.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
hwsrc=self.pg2.remote_mac,
pdst=self.pg0.remote_hosts[1].ip4,
psrc=self.pg0.local_ip4,
)
#
# Configure Proxy ARP for the subnet on PG0addresses on pg0
#
self.vapi.proxy_arp_add_del(
proxy={
"table_id": 0,
"low": self.pg0._local_ip4_subnet,
"hi": self.pg0._local_ip4_bcast,
},
is_add=1,
)
# Make pg2 un-numbered to pg0
#
self.pg2.set_unnumbered(self.pg0.sw_if_index)
#
# Enable pg2 for proxy ARP
#
self.pg2.set_proxy_arp()
#
# Send the ARP request with an originating address that
# is VPP's own address
#
rx = self.send_and_expect(self.pg2, [arp_req_from_me], self.pg2)
self.verify_arp_resp(
rx[0],
self.pg2.local_mac,
self.pg2.remote_mac,
self.pg0.remote_hosts[1].ip4,
self.pg0.local_ip4,
)
#
# validate we have not learned an ARP entry as a result of this
#
self.assertFalse(find_nbr(self, self.pg2.sw_if_index, self.pg0.local_ip4))
#
# setup a punt redirect so packets from the uplink go to the tap
#
redirect = VppIpPuntRedirect(
self, self.pg0.sw_if_index, self.pg2.sw_if_index, self.pg0.local_ip4
)
redirect.add_vpp_config()
p_tcp = (
Ether(
src=self.pg0.remote_mac,
dst=self.pg0.local_mac,
)
/ IP(src=self.pg0.remote_ip4, dst=self.pg0.local_ip4)
/ TCP(sport=80, dport=80)
/ Raw()
)
rx = self.send_and_expect(self.pg0, [p_tcp], self.pg2)
# there's no ARP entry so this is an ARP req
self.assertTrue(rx[0].haslayer(ARP))
# and ARP entry for VPP's pg0 address on the host interface
n1 = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_mac,
self.pg0.local_ip4,
is_no_fib_entry=True,
).add_vpp_config()
# now the packets shold forward
rx = self.send_and_expect(self.pg0, [p_tcp], self.pg2)
self.assertFalse(rx[0].haslayer(ARP))
self.assertEqual(rx[0][Ether].dst, self.pg2.remote_mac)
#
# flush the neighbor cache on the uplink
#
af = VppEnum.vl_api_address_family_t
self.vapi.ip_neighbor_flush(af.ADDRESS_IP4, self.pg0.sw_if_index)
# ensure we can still resolve the ARPs on the uplink
self.pg0.resolve_arp()
self.assertTrue(find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_ip4))
#
# cleanup
#
self.vapi.proxy_arp_add_del(
proxy={
"table_id": 0,
"low": self.pg0._local_ip4_subnet,
"hi": self.pg0._local_ip4_bcast,
},
is_add=0,
)
redirect.remove_vpp_config()
def test_proxy_arp(self):
"""Proxy ARP"""
self.pg1.generate_remote_hosts(2)
#
# Proxy ARP request packets for each interface
#
arp_req_pg0 = Ether(src=self.pg0.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
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",
hwsrc=self.pg1.remote_mac,
pdst="10.10.10.3",
psrc=self.pg1.remote_ip4,
)
arp_req_pg2 = Ether(src=self.pg2.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
hwsrc=self.pg2.remote_mac,
pdst="10.10.10.3",
psrc=self.pg1.remote_hosts[1].ip4,
)
arp_req_pg3 = Ether(src=self.pg3.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
hwsrc=self.pg3.remote_mac,
pdst="10.10.10.3",
psrc=self.pg3.remote_ip4,
)
#
# Configure Proxy ARP for 10.10.10.0 -> 10.10.10.124
#
self.vapi.proxy_arp_add_del(
proxy={"table_id": 0, "low": "10.10.10.2", "hi": "10.10.10.124"}, is_add=1
)
#
# No responses are sent when the interfaces are not enabled for proxy
# ARP
#
self.send_and_assert_no_replies(
self.pg0, arp_req_pg0, "ARP req from unconfigured interface"
)
self.send_and_assert_no_replies(
self.pg2, arp_req_pg2, "ARP req from unconfigured interface"
)
#
# Make pg2 un-numbered to pg1
# still won't reply.
#
self.pg2.set_unnumbered(self.pg1.sw_if_index)
self.send_and_assert_no_replies(
self.pg2, arp_req_pg2, "ARP req from unnumbered interface"
)
#
# Enable each interface to reply to proxy ARPs
#
for i in self.pg_interfaces:
i.set_proxy_arp()
#
# Now each of the interfaces should reply to a request to a proxied
# address
#
self.pg0.add_stream(arp_req_pg0)
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.pg0.remote_mac,
"10.10.10.3",
self.pg0.remote_ip4,
)
self.pg1.add_stream(arp_req_pg1)
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_mac,
"10.10.10.3",
self.pg1.remote_ip4,
)
self.pg2.add_stream(arp_req_pg2)
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,
"10.10.10.3",
self.pg1.remote_hosts[1].ip4,
)
#
# A request for an address out of the configured range
#
arp_req_pg1_hi = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
hwsrc=self.pg1.remote_mac,
pdst="10.10.10.125",
psrc=self.pg1.remote_ip4,
)
self.send_and_assert_no_replies(
self.pg1, arp_req_pg1_hi, "ARP req out of range HI"
)
arp_req_pg1_low = Ether(src=self.pg1.remote_mac, dst="ff:ff:ff:ff:ff:ff") / ARP(
op="who-has",
hwsrc=self.pg1.remote_mac,
pdst="10.10.10.1",
psrc=self.pg1.remote_ip4,
)
self.send_and_assert_no_replies(
self.pg1, arp_req_pg1_low, "ARP req out of range Low"
)
#
# Request for an address in the proxy range but from an interface
# in a different VRF
#
self.send_and_assert_no_replies(
self.pg3, arp_req_pg3, "ARP req from different VRF"
)
#
# Disable Each interface for proxy ARP
# - expect none to respond
#
for i in self.pg_interfaces:
i.set_proxy_arp(0)
self.send_and_assert_no_replies(self.pg0, arp_req_pg0, "ARP req from disable")
self.send_and_assert_no_replies(self.pg1, arp_req_pg1, "ARP req from disable")
self.send_and_assert_no_replies(self.pg2, arp_req_pg2, "ARP req from disable")
#
# clean up on interface 2
#
self.pg2.unset_unnumbered(self.pg1.sw_if_index)
def test_mpls(self):
"""MPLS"""
#
# Interface 2 does not yet have ip4 config
#
self.pg2.config_ip4()
self.pg2.generate_remote_hosts(2)
#
# Add a route with out going label via an ARP unresolved next-hop
#
ip_10_0_0_1 = VppIpRoute(
self,
"10.0.0.1",
32,
[
VppRoutePath(
self.pg2.remote_hosts[1].ip4, self.pg2.sw_if_index, labels=[55]
)
],
)
ip_10_0_0_1.add_vpp_config()
#
# packets should generate an ARP request
#
p = (
Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
/ IP(src=self.pg0.remote_ip4, dst="10.0.0.1")
/ UDP(sport=1234, dport=1234)
/ Raw(b"\xa5" * 100)
)
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg2.get_capture(1)
self.verify_arp_req(
rx[0], self.pg2.local_mac, self.pg2.local_ip4, self.pg2._remote_hosts[1].ip4
)
#
# now resolve the neighbours
#
self.pg2.configure_ipv4_neighbors()
#
# Now packet should be properly MPLS encapped.
# This verifies that MPLS link-type adjacencies are completed
# when the ARP entry resolves
#
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg2.get_capture(1)
self.verify_ip_o_mpls(
rx[0],
self.pg2.local_mac,
self.pg2.remote_hosts[1].mac,
55,
self.pg0.remote_ip4,
"10.0.0.1",
)
self.pg2.unconfig_ip4()
def test_arp_vrrp(self):
"""ARP reply with VRRP virtual src hw addr"""
#
# IP packet destined for pg1 remote host arrives on pg0 resulting
# in an ARP request for the address of the remote host on pg1
#
p0 = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
rx1 = self.send_and_expect(self.pg0, [p0], self.pg1)
self.verify_arp_req(
rx1[0], self.pg1.local_mac, self.pg1.local_ip4, self.pg1.remote_ip4
)
#
# ARP reply for address of pg1 remote host arrives on pg1 with
# the hw src addr set to a value in the VRRP IPv4 range of
# MAC addresses
#
p1 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / ARP(
op="is-at",
hwdst=self.pg1.local_mac,
hwsrc="00:00:5e:00:01:09",
pdst=self.pg1.local_ip4,
psrc=self.pg1.remote_ip4,
)
self.send_and_assert_no_replies(self.pg1, p1, "ARP reply")
#
# IP packet destined for pg1 remote host arrives on pg0 again.
# VPP should have an ARP entry for that address now and the packet
# should be sent out pg1.
#
rx1 = self.send_and_expect(self.pg0, [p0], self.pg1)
self.verify_ip(
rx1[0],
self.pg1.local_mac,
"00:00:5e:00:01:09",
self.pg0.remote_ip4,
self.pg1.remote_ip4,
)
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 should 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,
)
def test_arp_static(self):
"""ARP Static"""
self.pg2.generate_remote_hosts(3)
#
# Add a static ARP entry
#
static_arp = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[1].mac,
self.pg2.remote_hosts[1].ip4,
is_static=1,
)
static_arp.add_vpp_config()
#
# Add the connected prefix to the interface
#
self.pg2.config_ip4()
#
# We should now find the adj-fib
#
self.assertTrue(
find_nbr(
self, self.pg2.sw_if_index, self.pg2.remote_hosts[1].ip4, is_static=1
)
)
self.assertTrue(find_route(self, self.pg2.remote_hosts[1].ip4, 32))
#
# remove the connected
#
self.pg2.unconfig_ip4()
#
# put the interface into table 1
#
self.pg2.set_table_ip4(1)
#
# configure the same connected and expect to find the
# adj fib in the new table
#
self.pg2.config_ip4()
self.assertTrue(find_route(self, self.pg2.remote_hosts[1].ip4, 32, table_id=1))
#
# clean-up
#
self.pg2.unconfig_ip4()
static_arp.remove_vpp_config()
self.pg2.set_table_ip4(0)
def test_arp_static_replace_dynamic_same_mac(self):
"""ARP Static can replace Dynamic (same mac)"""
self.pg2.generate_remote_hosts(1)
dyn_arp = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].mac,
self.pg2.remote_hosts[0].ip4,
)
static_arp = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].mac,
self.pg2.remote_hosts[0].ip4,
is_static=1,
)
#
# Add a dynamic ARP entry
#
dyn_arp.add_vpp_config()
#
# We should find the dynamic nbr
#
self.assertFalse(
find_nbr(
self, self.pg2.sw_if_index, self.pg2.remote_hosts[0].ip4, is_static=1
)
)
self.assertTrue(
find_nbr(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].ip4,
is_static=0,
mac=self.pg2.remote_hosts[0].mac,
)
)
#
# Add a static ARP entry with the same mac
#
static_arp.add_vpp_config()
#
# We should now find the static nbr with the same mac
#
self.assertFalse(
find_nbr(
self, self.pg2.sw_if_index, self.pg2.remote_hosts[0].ip4, is_static=0
)
)
self.assertTrue(
find_nbr(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].ip4,
is_static=1,
mac=self.pg2.remote_hosts[0].mac,
)
)
#
# clean-up
#
static_arp.remove_vpp_config()
def test_arp_static_replace_dynamic_diff_mac(self):
"""ARP Static can replace Dynamic (diff mac)"""
self.pg2.generate_remote_hosts(2)
dyn_arp = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].mac,
self.pg2.remote_hosts[0].ip4,
)
static_arp = VppNeighbor(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[1].mac,
self.pg2.remote_hosts[0].ip4,
is_static=1,
)
#
# Add a dynamic ARP entry
#
dyn_arp.add_vpp_config()
#
# We should find the dynamic nbr
#
self.assertFalse(
find_nbr(
self, self.pg2.sw_if_index, self.pg2.remote_hosts[0].ip4, is_static=1
)
)
self.assertTrue(
find_nbr(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].ip4,
is_static=0,
mac=self.pg2.remote_hosts[0].mac,
)
)
#
# Add a static ARP entry with a changed mac
#
static_arp.add_vpp_config()
#
# We should now find the static nbr with a changed mac
#
self.assertFalse(
find_nbr(
self, self.pg2.sw_if_index, self.pg2.remote_hosts[0].ip4, is_static=0
)
)
self.assertTrue(
find_nbr(
self,
self.pg2.sw_if_index,
self.pg2.remote_hosts[0].ip4,
is_static=1,
mac=self.pg2.remote_hosts[1].mac,
)
)
#
# clean-up
#
static_arp.remove_vpp_config()
def test_arp_incomplete(self):
"""ARP Incomplete"""
self.pg1.generate_remote_hosts(4)
p0 = (
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()
)
p1 = (
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()
)
p2 = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst="1.1.1.1")
/ UDP(sport=1234, dport=1234)
/ Raw()
)
#
# a packet to an unresolved destination generates an ARP request
#
rx = self.send_and_expect(self.pg0, [p0], self.pg1)
self.verify_arp_req(
rx[0], self.pg1.local_mac, self.pg1.local_ip4, self.pg1._remote_hosts[1].ip4
)
#
# add a neighbour for remote host 1
#
static_arp = VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].mac,
self.pg1.remote_hosts[1].ip4,
is_static=1,
)
static_arp.add_vpp_config()
#
# add a route through remote host 3 hence we get an incomplete
#
VppIpRoute(
self,
"1.1.1.1",
32,
[VppRoutePath(self.pg1.remote_hosts[3].ip4, self.pg1.sw_if_index)],
).add_vpp_config()
rx = self.send_and_expect(self.pg0, [p2], self.pg1)
self.verify_arp_req(
rx[0], self.pg1.local_mac, self.pg1.local_ip4, self.pg1._remote_hosts[3].ip4
)
#
# change the interface's MAC
#
self.vapi.sw_interface_set_mac_address(
self.pg1.sw_if_index, "00:00:00:33:33:33"
)
#
# now ARP requests come from the new source mac
#
rx = self.send_and_expect(self.pg0, [p1], self.pg1)
self.verify_arp_req(
rx[0],
"00:00:00:33:33:33",
self.pg1.local_ip4,
self.pg1._remote_hosts[2].ip4,
)
rx = self.send_and_expect(self.pg0, [p2], self.pg1)
self.verify_arp_req(
rx[0],
"00:00:00:33:33:33",
self.pg1.local_ip4,
self.pg1._remote_hosts[3].ip4,
)
#
# packets to the resolved host also have the new source mac
#
rx = self.send_and_expect(self.pg0, [p0], self.pg1)
self.verify_ip(
rx[0],
"00:00:00:33:33:33",
self.pg1.remote_hosts[1].mac,
self.pg0.remote_ip4,
self.pg1.remote_hosts[1].ip4,
)
#
# set the mac address on the interface that does not have a
# configured subnet and thus no glean
#
self.vapi.sw_interface_set_mac_address(
self.pg2.sw_if_index, "00:00:00:33:33:33"
)
def test_garp(self):
"""GARP"""
#
# Generate some hosts on the LAN
#
self.pg1.generate_remote_hosts(4)
self.pg2.generate_remote_hosts(4)
#
# And an ARP entry
#
arp = VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].mac,
self.pg1.remote_hosts[1].ip4,
)
arp.add_vpp_config()
self.assertTrue(
find_nbr(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].ip4,
mac=self.pg1.remote_hosts[1].mac,
)
)
#
# Send a GARP (request) to swap the host 1's address to that of host 2
#
p1 = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1.remote_hosts[2].mac) / ARP(
op="who-has",
hwdst=self.pg1.local_mac,
hwsrc=self.pg1.remote_hosts[2].mac,
pdst=self.pg1.remote_hosts[1].ip4,
psrc=self.pg1.remote_hosts[1].ip4,
)
self.pg1.add_stream(p1)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertTrue(
find_nbr(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].ip4,
mac=self.pg1.remote_hosts[2].mac,
)
)
self.assert_equal(self.get_arp_rx_garp(self.pg1), 1)
#
# Send a GARP (reply) to swap the host 1's address to that of host 3
#
p1 = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1.remote_hosts[3].mac) / ARP(
op="is-at",
hwdst=self.pg1.local_mac,
hwsrc=self.pg1.remote_hosts[3].mac,
pdst=self.pg1.remote_hosts[1].ip4,
psrc=self.pg1.remote_hosts[1].ip4,
)
self.pg1.add_stream(p1)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertTrue(
find_nbr(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].ip4,
mac=self.pg1.remote_hosts[3].mac,
)
)
self.assert_equal(self.get_arp_rx_garp(self.pg1), 2)
#
# GARPs (request nor replies) for host we don't know yet
# don't result in new neighbour entries
#
p1 = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1.remote_hosts[3].mac) / ARP(
op="who-has",
hwdst=self.pg1.local_mac,
hwsrc=self.pg1.remote_hosts[3].mac,
pdst=self.pg1.remote_hosts[2].ip4,
psrc=self.pg1.remote_hosts[2].ip4,
)
self.pg1.add_stream(p1)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg1.remote_hosts[2].ip4)
)
p1 = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg1.remote_hosts[3].mac) / ARP(
op="is-at",
hwdst=self.pg1.local_mac,
hwsrc=self.pg1.remote_hosts[3].mac,
pdst=self.pg1.remote_hosts[2].ip4,
psrc=self.pg1.remote_hosts[2].ip4,
)
self.pg1.add_stream(p1)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg1.remote_hosts[2].ip4)
)
#
# IP address in different subnets are not learnt
#
self.pg2.configure_ipv4_neighbors()
cntr = self.statistics.get_err_counter(
"/err/arp-reply/l3_dst_address_not_local"
)
for op in ["is-at", "who-has"]:
p1 = [
(
Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_hosts[1].mac)
/ ARP(
op=op,
hwdst=self.pg2.local_mac,
hwsrc=self.pg2.remote_hosts[1].mac,
pdst=self.pg2.remote_hosts[1].ip4,
psrc=self.pg2.remote_hosts[1].ip4,
)
),
(
Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg2.remote_hosts[1].mac)
/ ARP(
op=op,
hwdst="ff:ff:ff:ff:ff:ff",
hwsrc=self.pg2.remote_hosts[1].mac,
pdst=self.pg2.remote_hosts[1].ip4,
psrc=self.pg2.remote_hosts[1].ip4,
)
),
]
self.send_and_assert_no_replies(self.pg1, p1)
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg2.remote_hosts[1].ip4)
)
# they are all dropped because the subnet's don't match
self.assertEqual(
cntr + 4,
self.statistics.get_err_counter("/err/arp-reply/l3_dst_address_not_local"),
)
def test_arp_incomplete2(self):
"""Incomplete Entries"""
#
# ensure that we throttle the ARP and ND requests
#
self.pg0.generate_remote_hosts(2)
#
# IPv4/ARP
#
ip_10_0_0_1 = VppIpRoute(
self,
"10.0.0.1",
32,
[VppRoutePath(self.pg0.remote_hosts[1].ip4, self.pg0.sw_if_index)],
)
ip_10_0_0_1.add_vpp_config()
p1 = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IP(src=self.pg1.remote_ip4, dst="10.0.0.1")
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg1.add_stream(p1 * 257)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg0._get_capture(1)
#
# how many we get is going to be dependent on the time for packet
# processing but it should be small
#
self.assertLess(len(rx), 64)
#
# IPv6/ND
#
ip_10_1 = VppIpRoute(
self,
"10::1",
128,
[
VppRoutePath(
self.pg0.remote_hosts[1].ip6,
self.pg0.sw_if_index,
proto=DpoProto.DPO_PROTO_IP6,
)
],
)
ip_10_1.add_vpp_config()
p1 = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_ip6, dst="10::1")
/ UDP(sport=1234, dport=1234)
/ Raw()
)
self.pg1.add_stream(p1 * 257)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
rx = self.pg0._get_capture(1)
#
# how many we get is going to be dependent on the time for packet
# processing but it should be small
#
self.assertLess(len(rx), 64)
def test_arp_forus(self):
"""ARP for for-us"""
#
# Test that VPP responds with ARP requests to addresses that
# are connected and local routes.
# Use one of the 'remote' addresses in the subnet as a local address
# The intention of this route is that it then acts like a secondary
# address added to an interface
#
self.pg0.generate_remote_hosts(2)
forus = VppIpRoute(
self,
self.pg0.remote_hosts[1].ip4,
32,
[
VppRoutePath(
"0.0.0.0",
self.pg0.sw_if_index,
type=FibPathType.FIB_PATH_TYPE_LOCAL,
)
],
)
forus.add_vpp_config()
p = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.pg0.remote_mac) / ARP(
op="who-has",
hwdst=self.pg0.local_mac,
hwsrc=self.pg0.remote_mac,
pdst=self.pg0.remote_hosts[1].ip4,
psrc=self.pg0.remote_ip4,
)
rx = self.send_and_expect(self.pg0, [p], self.pg0)
self.verify_arp_resp(
rx[0],
self.pg0.local_mac,
self.pg0.remote_mac,
self.pg0.remote_hosts[1].ip4,
self.pg0.remote_ip4,
)
def test_arp_table_swap(self):
#
# Generate some hosts on the LAN
#
N_NBRS = 4
self.pg1.generate_remote_hosts(N_NBRS)
for n in range(N_NBRS):
# a route thru each neighbour
VppIpRoute(
self,
"10.0.0.%d" % n,
32,
[VppRoutePath(self.pg1.remote_hosts[n].ip4, self.pg1.sw_if_index)],
).add_vpp_config()
# resolve each neighbour
p1 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / ARP(
op="is-at",
hwdst=self.pg1.local_mac,
hwsrc="00:00:5e:00:01:09",
pdst=self.pg1.local_ip4,
psrc=self.pg1.remote_hosts[n].ip4,
)
self.send_and_assert_no_replies(self.pg1, p1, "ARP reply")
self.logger.info(self.vapi.cli("sh ip neighbors"))
#
# swap the table pg1 is in
#
table = VppIpTable(self, 100).add_vpp_config()
self.pg1.unconfig_ip4()
self.pg1.set_table_ip4(100)
self.pg1.config_ip4()
#
# all neighbours are cleared
#
for n in range(N_NBRS):
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg1.remote_hosts[n].ip4)
)
#
# packets to all neighbours generate ARP requests
#
for n in range(N_NBRS):
# a route thru each neighbour
VppIpRoute(
self,
"10.0.0.%d" % n,
32,
[VppRoutePath(self.pg1.remote_hosts[n].ip4, self.pg1.sw_if_index)],
table_id=100,
).add_vpp_config()
p = (
Ether(src=self.pg1.remote_hosts[n].mac, dst=self.pg1.local_mac)
/ IP(src=self.pg1.remote_hosts[n].ip4, dst="10.0.0.%d" % n)
/ Raw(b"0x5" * 100)
)
rxs = self.send_and_expect(self.pg1, [p], self.pg1)
for rx in rxs:
self.verify_arp_req(
rx,
self.pg1.local_mac,
self.pg1.local_ip4,
self.pg1.remote_hosts[n].ip4,
)
self.pg1.unconfig_ip4()
self.pg1.set_table_ip4(0)
def test_glean_src_select(self):
"""Multi Connecteds"""
#
# configure multiple connected subnets on an interface
# and ensure that ARP requests for hosts on those subnets
# pick up the correct source address
#
conn1 = VppIpInterfaceAddress(self, self.pg1, "10.0.0.1", 24).add_vpp_config()
conn2 = VppIpInterfaceAddress(self, self.pg1, "10.0.1.1", 24).add_vpp_config()
p1 = (
Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
/ IP(src=self.pg1.remote_ip4, dst="10.0.0.128")
/ Raw(b"0x5" * 100)
)
rxs = self.send_and_expect(self.pg0, [p1], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.0.1", "10.0.0.128")
p2 = (
Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
/ IP(src=self.pg1.remote_ip4, dst="10.0.1.128")
/ Raw(b"0x5" * 100)
)
rxs = self.send_and_expect(self.pg0, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.1", "10.0.1.128")
#
# add a local address in the same subnet
# the source addresses are equivalent.
# VPP leaves the glean address being used for a prefix
# in place until that address is deleted.
#
conn3 = VppIpInterfaceAddress(self, self.pg1, "10.0.1.2", 24).add_vpp_config()
rxs = self.send_and_expect(self.pg0, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.1", "10.0.1.128")
#
# remove first address, which is currently in use
# the second address should be used now
#
conn2.remove_vpp_config()
rxs = self.send_and_expect(self.pg0, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.2", "10.0.1.128")
#
# add first address back. Second address should continue
# being used.
#
conn2 = VppIpInterfaceAddress(self, self.pg1, "10.0.1.1", 24).add_vpp_config()
rxs = self.send_and_expect(self.pg0, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.2", "10.0.1.128")
conn1.remove_vpp_config()
rxs = self.send_and_expect(self.pg0, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.2", "10.0.1.128")
# apply a connected prefix to an interface in a different table
VppIpRoute(
self,
"10.0.1.0",
24,
[VppRoutePath("0.0.0.0", self.pg1.sw_if_index)],
table_id=1,
).add_vpp_config()
rxs = self.send_and_expect(self.pg3, [p2], self.pg1)
for rx in rxs:
self.verify_arp_req(rx, self.pg1.local_mac, "10.0.1.2", "10.0.1.128")
# apply an attached prefix to the interface
# since there's no local address in this prefix,
# any other address is used
p3 = (
Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
/ IP(src=self.pg1.remote_ip4, dst="10.0.2.128")
/ Raw(b"0x5" * 100)
)
VppIpRoute(
self,
"10.0.2.0",
24,
[VppRoutePath("0.0.0.0", self.pg1.sw_if_index)],
).add_vpp_config()
rxs = self.send_and_expect(self.pg0, [p3], self.pg1)
for rx in rxs:
self.verify_arp_req(
rx, self.pg1.local_mac, self.pg1.local_ip4, "10.0.2.128"
)
# cleanup
conn3.remove_vpp_config()
conn2.remove_vpp_config()
@tag_fixme_vpp_workers
class NeighborStatsTestCase(VppTestCase):
"""ARP/ND Counters"""
@classmethod
def setUpClass(cls):
super(NeighborStatsTestCase, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(NeighborStatsTestCase, cls).tearDownClass()
def setUp(self):
super(NeighborStatsTestCase, self).setUp()
self.create_pg_interfaces(range(2))
# 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()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
super(NeighborStatsTestCase, self).tearDown()
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.admin_down()
def test_arp_stats(self):
"""ARP Counters"""
self.vapi.cli("adj counters enable")
self.pg1.generate_remote_hosts(2)
arp1 = VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[0].mac,
self.pg1.remote_hosts[0].ip4,
)
arp1.add_vpp_config()
arp2 = VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[1].mac,
self.pg1.remote_hosts[1].ip4,
)
arp2.add_vpp_config()
p1 = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_hosts[0].ip4)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
p2 = (
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()
)
rx = self.send_and_expect(self.pg0, p1 * NUM_PKTS, self.pg1)
rx = self.send_and_expect(self.pg0, p2 * NUM_PKTS, self.pg1)
self.assertEqual(NUM_PKTS, arp1.get_stats()["packets"])
self.assertEqual(NUM_PKTS, arp2.get_stats()["packets"])
rx = self.send_and_expect(self.pg0, p1 * NUM_PKTS, self.pg1)
self.assertEqual(NUM_PKTS * 2, arp1.get_stats()["packets"])
def test_nd_stats(self):
"""ND Counters"""
self.vapi.cli("adj counters enable")
self.pg0.generate_remote_hosts(3)
nd1 = VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[1].mac,
self.pg0.remote_hosts[1].ip6,
)
nd1.add_vpp_config()
nd2 = VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[2].mac,
self.pg0.remote_hosts[2].ip6,
)
nd2.add_vpp_config()
p1 = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_ip6, dst=self.pg0.remote_hosts[1].ip6)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
p2 = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_ip6, dst=self.pg0.remote_hosts[2].ip6)
/ UDP(sport=1234, dport=1234)
/ Raw()
)
rx = self.send_and_expect(self.pg1, p1 * 16, self.pg0)
rx = self.send_and_expect(self.pg1, p2 * 16, self.pg0)
self.assertEqual(16, nd1.get_stats()["packets"])
self.assertEqual(16, nd2.get_stats()["packets"])
rx = self.send_and_expect(self.pg1, p1 * NUM_PKTS, self.pg0)
self.assertEqual(NUM_PKTS + 16, nd1.get_stats()["packets"])
@tag_fixme_ubuntu2204
class NeighborAgeTestCase(VppTestCase):
"""ARP/ND Aging"""
@classmethod
def setUpClass(cls):
super(NeighborAgeTestCase, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(NeighborAgeTestCase, cls).tearDownClass()
def setUp(self):
super(NeighborAgeTestCase, self).setUp()
self.create_pg_interfaces(range(1))
# 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()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
super(NeighborAgeTestCase, self).tearDown()
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.admin_down()
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_ip_neighbor_config(self, af, max_number, max_age, recycle):
config = self.vapi.ip_neighbor_config_get(af)
self.assertEqual(config.af, af)
self.assertEqual(config.max_number, max_number)
self.assertEqual(config.max_age, max_age)
self.assertEqual(config.recycle, recycle)
def test_age(self):
"""Aging/Recycle"""
self.vapi.cli("set logging unthrottle 0")
self.vapi.cli("set logging size %d" % 0xFFFF)
self.pg0.generate_remote_hosts(201)
vaf = VppEnum.vl_api_address_family_t
#
# start listening on all interfaces
#
self.pg_enable_capture(self.pg_interfaces)
#
# Verify neighbor configuration defaults
#
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=50000, max_age=0, recycle=False
)
#
# Set the neighbor configuration:
# limi = 200
# age = 0 seconds
# recycle = false
#
self.vapi.ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=False
)
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=False
)
self.vapi.cli("sh ip neighbor-config")
# add the 198 neighbours that should pass (-1 for one created in setup)
for ii in range(200):
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[ii].mac,
self.pg0.remote_hosts[ii].ip4,
).add_vpp_config()
# one more neighbor over the limit should fail
with self.vapi.assert_negative_api_retval():
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[200].mac,
self.pg0.remote_hosts[200].ip4,
).add_vpp_config()
#
# change the config to allow recycling the old neighbors
#
self.vapi.ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=True
)
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=True
)
# now new additions are allowed
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[200].mac,
self.pg0.remote_hosts[200].ip4,
).add_vpp_config()
# add the first neighbor we configured has been re-used
self.assertFalse(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[0].ip4)
)
self.assertTrue(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[200].ip4)
)
#
# change the config to age old neighbors
#
self.vapi.ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=2, recycle=True
)
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=2, recycle=True
)
self.vapi.cli("sh ip4 neighbor-sorted")
# age out neighbors
self.virtual_sleep(3)
#
# expect probes from all these ARP entries as they age
# 3 probes for each neighbor 3*200 = 600
rxs = self.pg0.get_capture(600, timeout=2)
for ii in range(3):
for jj in range(200):
rx = rxs[ii * 200 + jj]
# rx.show()
#
# 3 probes sent then 1 more second to see if a reply comes, before
# they age out
#
self.virtual_sleep(1)
self.assertFalse(
self.vapi.ip_neighbor_dump(sw_if_index=0xFFFFFFFF, af=vaf.ADDRESS_IP4)
)
#
# load up some neighbours again with 2s aging enabled
# they should be removed after 10s (2s age + 4s for probes + gap)
# check for the add and remove events
#
enum = VppEnum.vl_api_ip_neighbor_event_flags_t
self.vapi.want_ip_neighbor_events_v2(enable=1)
for ii in range(10):
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[ii].mac,
self.pg0.remote_hosts[ii].ip4,
).add_vpp_config()
e = self.vapi.wait_for_event(1, "ip_neighbor_event_v2")
self.assertEqual(e.flags, enum.IP_NEIGHBOR_API_EVENT_FLAG_ADDED)
self.assertEqual(str(e.neighbor.ip_address), self.pg0.remote_hosts[ii].ip4)
self.assertEqual(e.neighbor.mac_address, self.pg0.remote_hosts[ii].mac)
self.virtual_sleep(10)
self.assertFalse(
self.vapi.ip_neighbor_dump(sw_if_index=0xFFFFFFFF, af=vaf.ADDRESS_IP4)
)
evs = []
for ii in range(10):
e = self.vapi.wait_for_event(1, "ip_neighbor_event_v2")
self.assertEqual(e.flags, enum.IP_NEIGHBOR_API_EVENT_FLAG_REMOVED)
evs.append(e)
# check we got the correct mac/ip pairs - done separately
# because we don't care about the order the remove notifications
# arrive
for ii in range(10):
found = False
mac = self.pg0.remote_hosts[ii].mac
ip = self.pg0.remote_hosts[ii].ip4
for e in evs:
if e.neighbor.mac_address == mac and str(e.neighbor.ip_address) == ip:
found = True
break
self.assertTrue(found)
#
# check if we can set age and recycle with empty neighbor list
#
self.vapi.ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=1000, recycle=True
)
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=1000, recycle=True
)
#
# load up some neighbours again, then disable the aging
# they should still be there in 10 seconds time
#
for ii in range(10):
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[ii].mac,
self.pg0.remote_hosts[ii].ip4,
).add_vpp_config()
self.vapi.ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=False
)
self.verify_ip_neighbor_config(
af=vaf.ADDRESS_IP4, max_number=200, max_age=0, recycle=False
)
self.virtual_sleep(10)
self.assertTrue(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[0].ip4)
)
class NeighborReplaceTestCase(VppTestCase):
"""ARP/ND Replacement"""
@classmethod
def setUpClass(cls):
super(NeighborReplaceTestCase, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(NeighborReplaceTestCase, cls).tearDownClass()
def setUp(self):
super(NeighborReplaceTestCase, self).setUp()
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()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
super(NeighborReplaceTestCase, self).tearDown()
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.admin_down()
def test_replace(self):
"""replace"""
N_HOSTS = 16
for i in self.pg_interfaces:
i.generate_remote_hosts(N_HOSTS)
i.configure_ipv4_neighbors()
i.configure_ipv6_neighbors()
# replace them all
self.vapi.ip_neighbor_replace_begin()
self.vapi.ip_neighbor_replace_end()
for i in self.pg_interfaces:
for h in range(N_HOSTS):
self.assertFalse(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[h].ip4)
)
self.assertFalse(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[h].ip6)
)
#
# and them all back via the API
#
for i in self.pg_interfaces:
for h in range(N_HOSTS):
VppNeighbor(
self, i.sw_if_index, i.remote_hosts[h].mac, i.remote_hosts[h].ip4
).add_vpp_config()
VppNeighbor(
self, i.sw_if_index, i.remote_hosts[h].mac, i.remote_hosts[h].ip6
).add_vpp_config()
#
# begin the replacement again, this time touch some
# the neighbours on pg1 so they are not deleted
#
self.vapi.ip_neighbor_replace_begin()
# update from the API all neighbours on pg1
for h in range(N_HOSTS):
VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[h].mac,
self.pg1.remote_hosts[h].ip4,
).add_vpp_config()
VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[h].mac,
self.pg1.remote_hosts[h].ip6,
).add_vpp_config()
# update from the data-plane all neighbours on pg3
self.pg3.configure_ipv4_neighbors()
self.pg3.configure_ipv6_neighbors()
# complete the replacement
self.logger.info(self.vapi.cli("sh ip neighbors"))
self.vapi.ip_neighbor_replace_end()
for i in self.pg_interfaces:
if i == self.pg1 or i == self.pg3:
# neighbours on pg1 and pg3 are still present
for h in range(N_HOSTS):
self.assertTrue(
find_nbr(self, i.sw_if_index, i.remote_hosts[h].ip4)
)
self.assertTrue(
find_nbr(self, i.sw_if_index, i.remote_hosts[h].ip6)
)
else:
# all other neighbours are toast
for h in range(N_HOSTS):
self.assertFalse(
find_nbr(self, i.sw_if_index, i.remote_hosts[h].ip4)
)
self.assertFalse(
find_nbr(self, i.sw_if_index, i.remote_hosts[h].ip6)
)
class NeighborFlush(VppTestCase):
"""Neighbor Flush"""
@classmethod
def setUpClass(cls):
super(NeighborFlush, cls).setUpClass()
@classmethod
def tearDownClass(cls):
super(NeighborFlush, cls).tearDownClass()
def setUp(self):
super(NeighborFlush, self).setUp()
self.create_pg_interfaces(range(2))
for i in self.pg_interfaces:
i.admin_up()
i.config_ip4()
i.config_ip6()
i.resolve_arp()
i.resolve_ndp()
def tearDown(self):
super(NeighborFlush, self).tearDown()
for i in self.pg_interfaces:
i.unconfig_ip4()
i.unconfig_ip6()
i.admin_down()
def test_flush(self):
"""Neighbour Flush"""
e = VppEnum
nf = e.vl_api_ip_neighbor_flags_t
af = e.vl_api_address_family_t
N_HOSTS = 16
static = [False, True]
self.pg0.generate_remote_hosts(N_HOSTS)
self.pg1.generate_remote_hosts(N_HOSTS)
for s in static:
# a few v4 and v6 dynamic neoghbors
for n in range(N_HOSTS):
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[n].mac,
self.pg0.remote_hosts[n].ip4,
is_static=s,
).add_vpp_config()
VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[n].mac,
self.pg1.remote_hosts[n].ip6,
is_static=s,
).add_vpp_config()
# flush the interfaces individually
self.vapi.ip_neighbor_flush(af.ADDRESS_IP4, self.pg0.sw_if_index)
# check we haven't flushed that which we shouldn't
for n in range(N_HOSTS):
self.assertTrue(
find_nbr(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[n].ip6,
is_static=s,
)
)
self.vapi.ip_neighbor_flush(af.ADDRESS_IP6, self.pg1.sw_if_index)
for n in range(N_HOSTS):
self.assertFalse(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[n].ip4)
)
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg1.remote_hosts[n].ip6)
)
# add the nieghbours back
for n in range(N_HOSTS):
VppNeighbor(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[n].mac,
self.pg0.remote_hosts[n].ip4,
is_static=s,
).add_vpp_config()
VppNeighbor(
self,
self.pg1.sw_if_index,
self.pg1.remote_hosts[n].mac,
self.pg1.remote_hosts[n].ip6,
is_static=s,
).add_vpp_config()
self.logger.info(self.vapi.cli("sh ip neighbor"))
# flush both interfaces at the same time
self.vapi.ip_neighbor_flush(af.ADDRESS_IP6, 0xFFFFFFFF)
# check we haven't flushed that which we shouldn't
for n in range(N_HOSTS):
self.assertTrue(
find_nbr(
self,
self.pg0.sw_if_index,
self.pg0.remote_hosts[n].ip4,
is_static=s,
)
)
self.vapi.ip_neighbor_flush(af.ADDRESS_IP4, 0xFFFFFFFF)
for n in range(N_HOSTS):
self.assertFalse(
find_nbr(self, self.pg0.sw_if_index, self.pg0.remote_hosts[n].ip4)
)
self.assertFalse(
find_nbr(self, self.pg1.sw_if_index, self.pg1.remote_hosts[n].ip6)
)
if __name__ == "__main__":
unittest.main(testRunner=VppTestRunner)
Reference in New Issue View Git Blame Copy Permalink