src/plugins/mss_clamp/test/test_mss_clamp.py

   1 #!/usr/bin/env python3
   2
   3 import unittest
   4
   5 from framework import VppTestCase, VppTestRunner
   6
   7 from scapy.layers.inet import IP, TCP
   8 from scapy.layers.inet6 import IPv6
   9 from scapy.layers.l2 import Ether
  10 from scapy.packet import Raw
  11
  12
  13 class TestMSSClamp(VppTestCase):
  14     """ TCP MSS Clamping Test Case """
  15
  16     def setUp(self):
  17         super(TestMSSClamp, self).setUp()
  18
  19         # create 2 pg interfaces
  20         self.create_pg_interfaces(range(2))
  21
  22         for i in self.pg_interfaces:
  23             i.admin_up()
  24             i.config_ip4()
  25             i.resolve_arp()
  26             i.config_ip6()
  27             i.resolve_ndp()
  28
  29     def tearDown(self):
  30         for i in self.pg_interfaces:
  31             i.unconfig_ip4()
  32             i.unconfig_ip6()
  33             i.admin_down()
  34         super(TestMSSClamp, self).tearDown()
  35
  36     def verify_pkt(self, rx, expected_mss):
  37         # check that the MSS size equals the expected value
  38         # and the IP and TCP checksums are correct
  39         tcp = rx[TCP]
  40         tcp_csum = tcp.chksum
  41         del tcp.chksum
  42         ip_csum = 0
  43         if (rx.haslayer(IP)):
  44             ip_csum = rx[IP].chksum
  45             del rx[IP].chksum
  46
  47         opt = tcp.options
  48         self.assertEqual(opt[0][0], 'MSS')
  49         self.assertEqual(opt[0][1], expected_mss)
  50         # recalculate checksums
  51         rx = rx.__class__(bytes(rx))
  52         tcp = rx[TCP]
  53         self.assertEqual(tcp_csum, tcp.chksum)
  54         if (rx.haslayer(IP)):
  55             self.assertEqual(ip_csum, rx[IP].chksum)
  56
  57     def send_and_verify_ip4(self, src_pg, dst_pg, mss, expected_mss):
  58         # IPv4 TCP packet with the requested MSS option.
  59         # from a host on src_pg to a host on dst_pg.
  60         p = (Ether(dst=src_pg.local_mac,
  61                    src=src_pg.remote_mac) /
  62              IP(src=src_pg.remote_ip4,
  63                 dst=dst_pg.remote_ip4) /
  64              TCP(sport=1234, dport=1234,
  65                  flags="S",
  66                  options=[('MSS', (mss)), ('EOL', None)]) /
  67              Raw('\xa5' * 100))
  68
  69         rxs = self.send_and_expect(src_pg, p * 65, dst_pg)
  70
  71         for rx in rxs:
  72             self.verify_pkt(rx, expected_mss)
  73
  74     def send_and_verify_ip6(self, src_pg, dst_pg, mss, expected_mss):
  75         #
  76         # IPv6 TCP packet with the requested MSS option.
  77         # from a host on src_pg to a host on dst_pg.
  78         #
  79         p = (Ether(dst=src_pg.local_mac,
  80                    src=src_pg.remote_mac) /
  81              IPv6(src=src_pg.remote_ip6,
  82                   dst=dst_pg.remote_ip6) /
  83              TCP(sport=1234, dport=1234,
  84                  flags="S",
  85                  options=[('MSS', (mss)), ('EOL', None)]) /
  86              Raw('\xa5' * 100))
  87
  88         rxs = self.send_and_expect(src_pg, p * 65, dst_pg)
  89
  90         for rx in rxs:
  91             self.verify_pkt(rx, expected_mss)
  92
  93     def test_tcp_mss_clamping_ip4_tx(self):
  94         """ IP4 TCP MSS Clamping TX """
  95
  96         # enable the TCP MSS clamping feature to lower the MSS to 1424.
  97         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
  98                                            ipv4_mss=1424, ipv6_mss=0,
  99                                            ipv4_direction=3, ipv6_direction=0)
 100
 101         # Verify that the feature is enabled.
 102         rv, reply = self.vapi.mss_clamp_get(sw_if_index=self.pg1.sw_if_index)
 103         self.assertEqual(reply[0].ipv4_mss, 1424)
 104         self.assertEqual(reply[0].ipv4_direction, 3)
 105
 106         # Send syn packets and verify that the MSS value is lowered.
 107         self.send_and_verify_ip4(self.pg0, self.pg1, 1460, 1424)
 108
 109         # check the stats
 110         stats = self.statistics.get_counter(
 111             '/err/tcp-mss-clamping-ip4-out/clamped')
 112         self.assertEqual(sum(stats), 65)
 113
 114         # Send syn packets with small enough MSS values and verify they are
 115         # unchanged.
 116         self.send_and_verify_ip4(self.pg0, self.pg1, 1400, 1400)
 117
 118         # enable the the feature only in TX direction
 119         # and change the max MSS value
 120         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 121                                            ipv4_mss=1420, ipv6_mss=0,
 122                                            ipv4_direction=2, ipv6_direction=0)
 123
 124         # Send syn packets and verify that the MSS value is lowered.
 125         self.send_and_verify_ip4(self.pg0, self.pg1, 1460, 1420)
 126
 127         # enable the the feature only in RX direction
 128         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 129                                            ipv4_mss=1424, ipv6_mss=0,
 130                                            ipv4_direction=1, ipv6_direction=0)
 131
 132         # Send the packets again and ensure they are unchanged.
 133         self.send_and_verify_ip4(self.pg0, self.pg1, 1460, 1460)
 134
 135         # disable the feature
 136         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 137                                            ipv4_mss=0, ipv6_mss=0,
 138                                            ipv4_direction=0, ipv6_direction=0)
 139
 140         # Send the packets again and ensure they are unchanged.
 141         self.send_and_verify_ip4(self.pg0, self.pg1, 1460, 1460)
 142
 143     def test_tcp_mss_clamping_ip4_rx(self):
 144         """ IP4 TCP MSS Clamping RX """
 145
 146         # enable the TCP MSS clamping feature to lower the MSS to 1424.
 147         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 148                                            ipv4_mss=1424, ipv6_mss=0,
 149                                            ipv4_direction=3, ipv6_direction=0)
 150
 151         # Verify that the feature is enabled.
 152         rv, reply = self.vapi.mss_clamp_get(sw_if_index=self.pg1.sw_if_index)
 153         self.assertEqual(reply[0].ipv4_mss, 1424)
 154         self.assertEqual(reply[0].ipv4_direction, 3)
 155
 156         # Send syn packets and verify that the MSS value is lowered.
 157         self.send_and_verify_ip4(self.pg1, self.pg0, 1460, 1424)
 158
 159         # check the stats
 160         stats = self.statistics.get_counter(
 161             '/err/tcp-mss-clamping-ip4-in/clamped')
 162         self.assertEqual(sum(stats), 65)
 163
 164         # Send syn packets with small enough MSS values and verify they are
 165         # unchanged.
 166         self.send_and_verify_ip4(self.pg1, self.pg0, 1400, 1400)
 167
 168         # enable the the feature only in RX direction
 169         # and change the max MSS value
 170         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 171                                            ipv4_mss=1420, ipv6_mss=0,
 172                                            ipv4_direction=1, ipv6_direction=0)
 173
 174         # Send syn packets and verify that the MSS value is lowered.
 175         self.send_and_verify_ip4(self.pg1, self.pg0, 1460, 1420)
 176
 177         # enable the the feature only in TX direction
 178         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 179                                            ipv4_mss=1424, ipv6_mss=0,
 180                                            ipv4_direction=2, ipv6_direction=0)
 181
 182         # Send the packets again and ensure they are unchanged.
 183         self.send_and_verify_ip4(self.pg1, self.pg0, 1460, 1460)
 184
 185         # disable the feature
 186         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 187                                            ipv4_mss=0, ipv6_mss=0,
 188                                            ipv4_direction=0, ipv6_direction=0)
 189
 190         # Send the packets again and ensure they are unchanged.
 191         self.send_and_verify_ip4(self.pg1, self.pg0, 1460, 1460)
 192
 193     def test_tcp_mss_clamping_ip6_tx(self):
 194         """ IP6 TCP MSS Clamping TX """
 195
 196         # enable the TCP MSS clamping feature to lower the MSS to 1424.
 197         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 198                                            ipv4_mss=0, ipv6_mss=1424,
 199                                            ipv4_direction=0, ipv6_direction=3)
 200
 201         # Verify that the feature is enabled.
 202         rv, reply = self.vapi.mss_clamp_get(sw_if_index=self.pg1.sw_if_index)
 203         self.assertEqual(reply[0].ipv6_mss, 1424)
 204         self.assertEqual(reply[0].ipv6_direction, 3)
 205
 206         # Send syn packets and verify that the MSS value is lowered.
 207         self.send_and_verify_ip6(self.pg0, self.pg1, 1460, 1424)
 208
 209         # check the stats
 210         stats = self.statistics.get_counter(
 211             '/err/tcp-mss-clamping-ip6-out/clamped')
 212         self.assertEqual(sum(stats), 65)
 213
 214         # Send syn packets with small enough MSS values and verify they are
 215         # unchanged.
 216         self.send_and_verify_ip6(self.pg0, self.pg1, 1400, 1400)
 217
 218         # enable the the feature only in TX direction
 219         # and change the max MSS value
 220         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 221                                            ipv4_mss=0, ipv6_mss=1420,
 222                                            ipv4_direction=0, ipv6_direction=2)
 223
 224         # Send syn packets and verify that the MSS value is lowered.
 225         self.send_and_verify_ip6(self.pg0, self.pg1, 1460, 1420)
 226
 227         # enable the the feature only in RX direction
 228         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 229                                            ipv4_mss=0, ipv6_mss=1424,
 230                                            ipv4_direction=0, ipv6_direction=1)
 231
 232         # Send the packets again and ensure they are unchanged.
 233         self.send_and_verify_ip6(self.pg0, self.pg1, 1460, 1460)
 234
 235         # disable the feature
 236         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 237                                            ipv4_mss=0, ipv6_mss=0,
 238                                            ipv4_direction=0, ipv6_direction=0)
 239
 240         # Send the packets again and ensure they are unchanged.
 241         self.send_and_verify_ip6(self.pg0, self.pg1, 1460, 1460)
 242
 243     def test_tcp_mss_clamping_ip6_rx(self):
 244         """ IP6 TCP MSS Clamping RX """
 245
 246         # enable the TCP MSS clamping feature to lower the MSS to 1424.
 247         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 248                                            ipv4_mss=0, ipv6_mss=1424,
 249                                            ipv4_direction=0, ipv6_direction=3)
 250
 251         # Verify that the feature is enabled.
 252         rv, reply = self.vapi.mss_clamp_get(sw_if_index=self.pg1.sw_if_index)
 253         self.assertEqual(reply[0].ipv6_mss, 1424)
 254         self.assertEqual(reply[0].ipv6_direction, 3)
 255
 256         # Send syn packets and verify that the MSS value is lowered.
 257         self.send_and_verify_ip6(self.pg1, self.pg0, 1460, 1424)
 258
 259         # check the stats
 260         stats = self.statistics.get_counter(
 261             '/err/tcp-mss-clamping-ip6-in/clamped')
 262         self.assertEqual(sum(stats), 65)
 263
 264         # Send syn packets with small enough MSS values and verify they are
 265         # unchanged.
 266         self.send_and_verify_ip6(self.pg1, self.pg0, 1400, 1400)
 267
 268         # enable the the feature only in RX direction
 269         # and change the max MSS value
 270         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 271                                            ipv4_mss=0, ipv6_mss=1420,
 272                                            ipv4_direction=0, ipv6_direction=1)
 273
 274         # Send syn packets and verify that the MSS value is lowered.
 275         self.send_and_verify_ip6(self.pg1, self.pg0, 1460, 1420)
 276
 277         # enable the the feature only in TX direction
 278         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 279                                            ipv4_mss=0, ipv6_mss=1424,
 280                                            ipv4_direction=0, ipv6_direction=2)
 281
 282         # Send the packets again and ensure they are unchanged.
 283         self.send_and_verify_ip6(self.pg1, self.pg0, 1460, 1460)
 284
 285         # disable the feature
 286         self.vapi.mss_clamp_enable_disable(self.pg1.sw_if_index,
 287                                            ipv4_mss=0, ipv6_mss=0,
 288                                            ipv4_direction=0, ipv6_direction=0)
 289
 290         # Send the packets again and ensure they are unchanged.
 291         self.send_and_verify_ip6(self.pg1, self.pg0, 1460, 1460)
 292
 293
 294 if __name__ == '__main__':
 295     unittest.main(testRunner=VppTestRunner)