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