test/test_mtu.py

   1 #!/usr/bin/env python3
   2 """IP4 and IP6 MTU functional tests"""
   3
   4 #
   5 # Add tests for:
   6 # - sub interfaces
   7 # - Verify that adjacencies inherit MTU correctly
   8 # - Verify that sub-interfaces inherit MTU correctly
   9 # - Different types of interfaces?
  10 #
  11 import unittest
  12 from scapy.layers.inet6 import IPv6, Ether, IP, UDP, ICMPv6PacketTooBig
  13 from scapy.layers.inet import ICMP
  14 from framework import VppTestCase, VppTestRunner
  15 from vpp_ip import DpoProto
  16 from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathProto
  17 from socket import AF_INET, AF_INET6, inet_pton
  18 from util import reassemble4
  19
  20
  21 """ Test_mtu is a subclass of VPPTestCase classes.
  22     MTU tests.
  23 """
  24
  25
  26 class TestMTU(VppTestCase):
  27     """MTU Test Case"""
  28
  29     maxDiff = None
  30
  31     @classmethod
  32     def setUpClass(cls):
  33         super(TestMTU, cls).setUpClass()
  34         cls.create_pg_interfaces(range(2))
  35         cls.interfaces = list(cls.pg_interfaces)
  36
  37     @classmethod
  38     def tearDownClass(cls):
  39         super(TestMTU, cls).tearDownClass()
  40
  41     def setUp(self):
  42         super(TestMTU, self).setUp()
  43         for i in self.interfaces:
  44             i.admin_up()
  45             i.config_ip4()
  46             i.config_ip6()
  47             i.disable_ipv6_ra()
  48             i.resolve_arp()
  49             i.resolve_ndp()
  50
  51     def tearDown(self):
  52         super(TestMTU, self).tearDown()
  53         if not self.vpp_dead:
  54             for i in self.pg_interfaces:
  55                 i.unconfig_ip4()
  56                 i.unconfig_ip6()
  57                 i.admin_down()
  58
  59     def validate(self, rx, expected):
  60         self.assertEqual(rx, expected.__class__(expected))
  61
  62     def validate_bytes(self, rx, expected):
  63         self.assertEqual(rx, expected)
  64
  65     def payload(self, len):
  66         return "x" * len
  67
  68     def get_mtu(self, sw_if_index):
  69         rv = self.vapi.sw_interface_dump(sw_if_index=sw_if_index)
  70         for i in rv:
  71             if i.sw_if_index == sw_if_index:
  72                 return i.mtu[0]
  73         return 0
  74
  75     def test_ip4_mtu(self):
  76         """IP4 MTU test"""
  77
  78         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
  79         p_ip4 = IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, flags="DF")
  80
  81         current_mtu = self.get_mtu(self.pg1.sw_if_index)
  82
  83         p_payload = UDP(sport=1234, dport=1234) / self.payload(current_mtu - 20 - 8)
  84
  85         p4 = p_ether / p_ip4 / p_payload
  86         p4_reply = p_ip4 / p_payload
  87         p4_reply.ttl -= 1
  88         rx = self.send_and_expect(self.pg0, p4 * 11, self.pg1)
  89         for p in rx:
  90             self.validate(p[1], p4_reply)
  91
  92         # MTU
  93         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [576, 0, 0, 0])
  94         self.assertEqual(576, self.get_mtu(self.pg1.sw_if_index))
  95
  96         # Should fail. Too large MTU
  97         p_icmp4 = ICMP(
  98             type="dest-unreach",
  99             code="fragmentation-needed",
 100             nexthopmtu=576,
 101             chksum=0x2DBB,
 102         )
 103         icmp4_reply = (
 104             IP(src=self.pg0.local_ip4, dst=self.pg0.remote_ip4, ttl=254, len=576, id=0)
 105             / p_icmp4
 106             / p_ip4
 107             / p_payload
 108         )
 109         n = icmp4_reply.__class__(icmp4_reply)
 110         s = bytes(icmp4_reply)
 111         icmp4_reply = s[0:576]
 112         rx = self.send_and_expect_some(self.pg0, p4 * 11, self.pg0)
 113         for p in rx:
 114             # p.show2()
 115             # n.show2()
 116             self.validate_bytes(bytes(p[1]), icmp4_reply)
 117
 118         # Now with DF off. Expect fragments.
 119         # First go with 1500 byte packets.
 120         p_payload = UDP(sport=1234, dport=1234) / self.payload(1500 - 20 - 8)
 121         p4 = p_ether / p_ip4 / p_payload
 122         p4.flags = 0
 123         p4_reply = p_ip4 / p_payload
 124         p4_reply.ttl = p_ip4.ttl - 1
 125         p4_reply.flags = 0
 126         p4_reply.id = 256
 127         self.pg_enable_capture()
 128         self.pg0.add_stream(p4 * 1)
 129         self.pg_start()
 130         rx = self.pg1.get_capture(3)
 131         reass_pkt = reassemble4(rx)
 132         self.validate(reass_pkt, p4_reply)
 133
 134         """
 135         # Now what happens with a 9K frame
 136         p_payload = UDP(sport=1234, dport=1234) / self.payload(
 137             current_mtu - 20 - 8)
 138         p4 = p_ether / p_ip4 / p_payload
 139         p4.flags = 0
 140         p4_reply = p_ip4 / p_payload
 141         p4_reply.ttl = 62 # check this
 142         p4_reply.flags = 0
 143         p4_reply.id = 512
 144
 145         self.pg_enable_capture()
 146         self.pg0.add_stream(p4*1)
 147         self.pg_start()
 148         rx = self.pg1.get_capture(16)
 149         reass_pkt = reassemble4(rx)
 150         reass_pkt.show2()
 151         p4_reply.show2()
 152         self.validate(reass_pkt, p4_reply)
 153         """
 154
 155         # Reset MTU
 156         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [current_mtu, 0, 0, 0])
 157
 158     def test_ip6_mtu(self):
 159         """IP6 MTU test"""
 160
 161         current_mtu = self.get_mtu(self.pg1.sw_if_index)
 162
 163         p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
 164         p_ip6 = IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6)
 165
 166         p_payload = UDP(sport=1234, dport=1234) / self.payload(current_mtu - 40 - 8)
 167
 168         p6 = p_ether / p_ip6 / p_payload
 169         p6_reply = p_ip6 / p_payload
 170         p6_reply.hlim -= 1
 171         rx = self.send_and_expect(self.pg0, p6 * 9, self.pg1)
 172         for p in rx:
 173             self.validate(p[1], p6_reply)
 174
 175         # MTU (only checked on encap)
 176         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [1280, 0, 0, 0])
 177         self.assertEqual(1280, self.get_mtu(self.pg1.sw_if_index))
 178
 179         # Should fail. Too large MTU
 180         p_icmp6 = ICMPv6PacketTooBig(mtu=1280, cksum=0x4C7A)
 181         icmp6_reply = (
 182             IPv6(src=self.pg0.local_ip6, dst=self.pg0.remote_ip6, hlim=255, plen=1240)
 183             / p_icmp6
 184             / p_ip6
 185             / p_payload
 186         )
 187         icmp6_reply[2].hlim -= 1
 188         n = icmp6_reply.__class__(icmp6_reply)
 189         s = bytes(icmp6_reply)
 190         icmp6_reply_str = s[0:1280]
 191
 192         rx = self.send_and_expect_some(self.pg0, p6 * 9, self.pg0)
 193         for p in rx:
 194             self.validate_bytes(bytes(p[1]), icmp6_reply_str)
 195
 196         # Reset MTU
 197         self.vapi.sw_interface_set_mtu(self.pg1.sw_if_index, [current_mtu, 0, 0, 0])
 198
 199
 200 if __name__ == "__main__":
 201     unittest.main(testRunner=VppTestRunner)