test/test_mss_clamp.py

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