test/test_snat.py

   1 #!/usr/bin/env python
   2
   3 import socket
   4 import unittest
   5
   6 from framework import VppTestCase, VppTestRunner
   7
   8 from scapy.layers.inet import IP, TCP, UDP, ICMP
   9 from scapy.layers.l2 import Ether
  10 from util import ppp
  11
  12
  13 class TestSNAT(VppTestCase):
  14     """ SNAT Test Cases """
  15
  16     @classmethod
  17     def setUpClass(cls):
  18         super(TestSNAT, cls).setUpClass()
  19
  20         try:
  21             cls.tcp_port_in = 6303
  22             cls.tcp_port_out = 6303
  23             cls.udp_port_in = 6304
  24             cls.udp_port_out = 6304
  25             cls.icmp_id_in = 6305
  26             cls.icmp_id_out = 6305
  27             cls.snat_addr = '10.0.0.3'
  28
  29             cls.create_pg_interfaces(range(7))
  30             cls.interfaces = list(cls.pg_interfaces[0:4])
  31
  32             for i in cls.interfaces:
  33                 i.admin_up()
  34                 i.config_ip4()
  35                 i.resolve_arp()
  36
  37             cls.pg0.generate_remote_hosts(2)
  38             cls.pg0.configure_ipv4_neighbors()
  39
  40             cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
  41
  42             for i in cls.overlapping_interfaces:
  43                 i._local_ip4 = "172.16.255.1"
  44                 i._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
  45                 i._remote_hosts[0]._ip4 = "172.16.255.2"
  46                 i.set_table_ip4(i.sw_if_index)
  47                 i.config_ip4()
  48                 i.admin_up()
  49                 i.resolve_arp()
  50
  51         except Exception:
  52             super(TestSNAT, cls).tearDownClass()
  53             raise
  54
  55     def create_stream_in(self, in_if, out_if):
  56         """
  57         Create packet stream for inside network
  58
  59         :param in_if: Inside interface
  60         :param out_if: Outside interface
  61         """
  62         pkts = []
  63         # TCP
  64         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  65              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
  66              TCP(sport=self.tcp_port_in))
  67         pkts.append(p)
  68
  69         # UDP
  70         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  71              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
  72              UDP(sport=self.udp_port_in))
  73         pkts.append(p)
  74
  75         # ICMP
  76         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  77              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
  78              ICMP(id=self.icmp_id_in, type='echo-request'))
  79         pkts.append(p)
  80
  81         return pkts
  82
  83     def create_stream_out(self, out_if, dst_ip=None):
  84         """
  85         Create packet stream for outside network
  86
  87         :param out_if: Outside interface
  88         :param dst_ip: Destination IP address (Default use global SNAT address)
  89         """
  90         if dst_ip is None:
  91             dst_ip = self.snat_addr
  92         pkts = []
  93         # TCP
  94         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
  95              IP(src=out_if.remote_ip4, dst=dst_ip) /
  96              TCP(dport=self.tcp_port_out))
  97         pkts.append(p)
  98
  99         # UDP
 100         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 101              IP(src=out_if.remote_ip4, dst=dst_ip) /
 102              UDP(dport=self.udp_port_out))
 103         pkts.append(p)
 104
 105         # ICMP
 106         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 107              IP(src=out_if.remote_ip4, dst=dst_ip) /
 108              ICMP(id=self.icmp_id_out, type='echo-reply'))
 109         pkts.append(p)
 110
 111         return pkts
 112
 113     def verify_capture_out(self, capture, nat_ip=None, same_port=False,
 114                            packet_num=3):
 115         """
 116         Verify captured packets on outside network
 117
 118         :param capture: Captured packets
 119         :param nat_ip: Translated IP address (Default use global SNAT address)
 120         :param same_port: Sorce port number is not translated (Default False)
 121         :param packet_num: Expected number of packets (Default 3)
 122         """
 123         if nat_ip is None:
 124             nat_ip = self.snat_addr
 125         self.assertEqual(packet_num, len(capture))
 126         for packet in capture:
 127             try:
 128                 self.assertEqual(packet[IP].src, nat_ip)
 129                 if packet.haslayer(TCP):
 130                     if same_port:
 131                         self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 132                     else:
 133                         self.assertNotEqual(
 134                             packet[TCP].sport, self.tcp_port_in)
 135                     self.tcp_port_out = packet[TCP].sport
 136                 elif packet.haslayer(UDP):
 137                     if same_port:
 138                         self.assertEqual(packet[UDP].sport, self.udp_port_in)
 139                     else:
 140                         self.assertNotEqual(
 141                             packet[UDP].sport, self.udp_port_in)
 142                     self.udp_port_out = packet[UDP].sport
 143                 else:
 144                     if same_port:
 145                         self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 146                     else:
 147                         self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
 148                     self.icmp_id_out = packet[ICMP].id
 149             except:
 150                 self.logger.error(ppp("Unexpected or invalid packet "
 151                                       "(outside network):", packet))
 152                 raise
 153
 154     def verify_capture_in(self, capture, in_if, packet_num=3):
 155         """
 156         Verify captured packets on inside network
 157
 158         :param capture: Captured packets
 159         :param in_if: Inside interface
 160         :param packet_num: Expected number of packets (Default 3)
 161         """
 162         self.assertEqual(packet_num, len(capture))
 163         for packet in capture:
 164             try:
 165                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 166                 if packet.haslayer(TCP):
 167                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 168                 elif packet.haslayer(UDP):
 169                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 170                 else:
 171                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 172             except:
 173                 self.logger.error(ppp("Unexpected or invalid packet "
 174                                       "(inside network):", packet))
 175                 raise
 176
 177     def clear_snat(self):
 178         """
 179         Clear SNAT configuration.
 180         """
 181         interfaces = self.vapi.snat_interface_dump()
 182         for intf in interfaces:
 183             self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
 184                                                      intf.is_inside,
 185                                                      is_add=0)
 186
 187         static_mappings = self.vapi.snat_static_mapping_dump()
 188         for sm in static_mappings:
 189             self.vapi.snat_add_static_mapping(sm.local_ip_address,
 190                                               sm.external_ip_address,
 191                                               local_port=sm.local_port,
 192                                               external_port=sm.external_port,
 193                                               addr_only=sm.addr_only,
 194                                               vrf_id=sm.vrf_id,
 195                                               is_add=0)
 196
 197         adresses = self.vapi.snat_address_dump()
 198         for addr in adresses:
 199             self.vapi.snat_add_address_range(addr.ip_address,
 200                                              addr.ip_address,
 201                                              is_add=0)
 202
 203     def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
 204                                 external_port=0, vrf_id=0, is_add=1):
 205         """
 206         Add/delete S-NAT static mapping
 207
 208         :param local_ip: Local IP address
 209         :param external_ip: External IP address
 210         :param local_port: Local port number (Optional)
 211         :param external_port: External port number (Optional)
 212         :param vrf_id: VRF ID (Default 0)
 213         :param is_add: 1 if add, 0 if delete (Default add)
 214         """
 215         addr_only = 1
 216         if local_port and external_port:
 217             addr_only = 0
 218         l_ip = socket.inet_pton(socket.AF_INET, local_ip)
 219         e_ip = socket.inet_pton(socket.AF_INET, external_ip)
 220         self.vapi.snat_add_static_mapping(
 221             l_ip,
 222             e_ip,
 223             local_port,
 224             external_port,
 225             addr_only,
 226             vrf_id,
 227             is_add)
 228
 229     def snat_add_address(self, ip, is_add=1):
 230         """
 231         Add/delete S-NAT address
 232
 233         :param ip: IP address
 234         :param is_add: 1 if add, 0 if delete (Default add)
 235         """
 236         snat_addr = socket.inet_pton(socket.AF_INET, ip)
 237         self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
 238
 239     def test_dynamic(self):
 240         """ SNAT dynamic translation test """
 241
 242         self.snat_add_address(self.snat_addr)
 243         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 244         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 245                                                  is_inside=0)
 246
 247         # in2out
 248         pkts = self.create_stream_in(self.pg0, self.pg1)
 249         self.pg0.add_stream(pkts)
 250         self.pg_enable_capture(self.pg_interfaces)
 251         self.pg_start()
 252         capture = self.pg1.get_capture(len(pkts))
 253         self.verify_capture_out(capture)
 254
 255         # out2in
 256         pkts = self.create_stream_out(self.pg1)
 257         self.pg1.add_stream(pkts)
 258         self.pg_enable_capture(self.pg_interfaces)
 259         self.pg_start()
 260         capture = self.pg0.get_capture(len(pkts))
 261         self.verify_capture_in(capture, self.pg0)
 262
 263     def test_static_in(self):
 264         """ SNAT 1:1 NAT initialized from inside network """
 265
 266         nat_ip = "10.0.0.10"
 267         self.tcp_port_out = 6303
 268         self.udp_port_out = 6304
 269         self.icmp_id_out = 6305
 270
 271         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 272         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 273         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 274                                                  is_inside=0)
 275
 276         # in2out
 277         pkts = self.create_stream_in(self.pg0, self.pg1)
 278         self.pg0.add_stream(pkts)
 279         self.pg_enable_capture(self.pg_interfaces)
 280         self.pg_start()
 281         capture = self.pg1.get_capture(len(pkts))
 282         self.verify_capture_out(capture, nat_ip, True)
 283
 284         # out2in
 285         pkts = self.create_stream_out(self.pg1, nat_ip)
 286         self.pg1.add_stream(pkts)
 287         self.pg_enable_capture(self.pg_interfaces)
 288         self.pg_start()
 289         capture = self.pg0.get_capture(len(pkts))
 290         self.verify_capture_in(capture, self.pg0)
 291
 292     def test_static_out(self):
 293         """ SNAT 1:1 NAT initialized from outside network """
 294
 295         nat_ip = "10.0.0.20"
 296         self.tcp_port_out = 6303
 297         self.udp_port_out = 6304
 298         self.icmp_id_out = 6305
 299
 300         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 301         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 302         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 303                                                  is_inside=0)
 304
 305         # out2in
 306         pkts = self.create_stream_out(self.pg1, nat_ip)
 307         self.pg1.add_stream(pkts)
 308         self.pg_enable_capture(self.pg_interfaces)
 309         self.pg_start()
 310         capture = self.pg0.get_capture(len(pkts))
 311         self.verify_capture_in(capture, self.pg0)
 312
 313         # in2out
 314         pkts = self.create_stream_in(self.pg0, self.pg1)
 315         self.pg0.add_stream(pkts)
 316         self.pg_enable_capture(self.pg_interfaces)
 317         self.pg_start()
 318         capture = self.pg1.get_capture(len(pkts))
 319         self.verify_capture_out(capture, nat_ip, True)
 320
 321     def test_static_with_port_in(self):
 322         """ SNAT 1:1 NAT with port initialized from inside network """
 323
 324         self.tcp_port_out = 3606
 325         self.udp_port_out = 3607
 326         self.icmp_id_out = 3608
 327
 328         self.snat_add_address(self.snat_addr)
 329         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 330                                      self.tcp_port_in, self.tcp_port_out)
 331         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 332                                      self.udp_port_in, self.udp_port_out)
 333         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 334                                      self.icmp_id_in, self.icmp_id_out)
 335         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 336         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 337                                                  is_inside=0)
 338
 339         # in2out
 340         pkts = self.create_stream_in(self.pg0, self.pg1)
 341         self.pg0.add_stream(pkts)
 342         self.pg_enable_capture(self.pg_interfaces)
 343         self.pg_start()
 344         capture = self.pg1.get_capture(len(pkts))
 345         self.verify_capture_out(capture)
 346
 347         # out2in
 348         pkts = self.create_stream_out(self.pg1)
 349         self.pg1.add_stream(pkts)
 350         self.pg_enable_capture(self.pg_interfaces)
 351         self.pg_start()
 352         capture = self.pg0.get_capture(len(pkts))
 353         self.verify_capture_in(capture, self.pg0)
 354
 355     def test_static_with_port_out(self):
 356         """ SNAT 1:1 NAT with port initialized from outside network """
 357
 358         self.tcp_port_out = 30606
 359         self.udp_port_out = 30607
 360         self.icmp_id_out = 30608
 361
 362         self.snat_add_address(self.snat_addr)
 363         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 364                                      self.tcp_port_in, self.tcp_port_out)
 365         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 366                                      self.udp_port_in, self.udp_port_out)
 367         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 368                                      self.icmp_id_in, self.icmp_id_out)
 369         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 370         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 371                                                  is_inside=0)
 372
 373         # out2in
 374         pkts = self.create_stream_out(self.pg1)
 375         self.pg1.add_stream(pkts)
 376         self.pg_enable_capture(self.pg_interfaces)
 377         self.pg_start()
 378         capture = self.pg0.get_capture(len(pkts))
 379         self.verify_capture_in(capture, self.pg0)
 380
 381         # in2out
 382         pkts = self.create_stream_in(self.pg0, self.pg1)
 383         self.pg0.add_stream(pkts)
 384         self.pg_enable_capture(self.pg_interfaces)
 385         self.pg_start()
 386         capture = self.pg1.get_capture(len(pkts))
 387         self.verify_capture_out(capture)
 388
 389     def test_static_vrf_aware(self):
 390         """ SNAT 1:1 NAT VRF awareness """
 391
 392         nat_ip1 = "10.0.0.30"
 393         nat_ip2 = "10.0.0.40"
 394         self.tcp_port_out = 6303
 395         self.udp_port_out = 6304
 396         self.icmp_id_out = 6305
 397
 398         self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
 399                                      vrf_id=self.pg4.sw_if_index)
 400         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
 401                                      vrf_id=self.pg4.sw_if_index)
 402         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 403                                                  is_inside=0)
 404         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 405         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 406
 407         # inside interface VRF match SNAT static mapping VRF
 408         pkts = self.create_stream_in(self.pg4, self.pg3)
 409         self.pg4.add_stream(pkts)
 410         self.pg_enable_capture(self.pg_interfaces)
 411         self.pg_start()
 412         capture = self.pg3.get_capture(len(pkts))
 413         self.verify_capture_out(capture, nat_ip1, True)
 414
 415         # inside interface VRF don't match SNAT static mapping VRF (packets
 416         # are dropped)
 417         pkts = self.create_stream_in(self.pg0, self.pg3)
 418         self.pg0.add_stream(pkts)
 419         self.pg_enable_capture(self.pg_interfaces)
 420         self.pg_start()
 421         self.pg3.assert_nothing_captured()
 422
 423     def test_multiple_inside_interfaces(self):
 424         """
 425         SNAT multiple inside interfaces with non-overlapping address space
 426         """
 427
 428         self.snat_add_address(self.snat_addr)
 429         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 430         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
 431         self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index)
 432         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 433                                                  is_inside=0)
 434
 435         # in2out 1st interface
 436         pkts = self.create_stream_in(self.pg0, self.pg3)
 437         self.pg0.add_stream(pkts)
 438         self.pg_enable_capture(self.pg_interfaces)
 439         self.pg_start()
 440         capture = self.pg3.get_capture(len(pkts))
 441         self.verify_capture_out(capture)
 442
 443         # out2in 1st interface
 444         pkts = self.create_stream_out(self.pg3)
 445         self.pg3.add_stream(pkts)
 446         self.pg_enable_capture(self.pg_interfaces)
 447         self.pg_start()
 448         capture = self.pg0.get_capture(len(pkts))
 449         self.verify_capture_in(capture, self.pg0)
 450
 451         # in2out 2nd interface
 452         pkts = self.create_stream_in(self.pg1, self.pg3)
 453         self.pg1.add_stream(pkts)
 454         self.pg_enable_capture(self.pg_interfaces)
 455         self.pg_start()
 456         capture = self.pg3.get_capture(len(pkts))
 457         self.verify_capture_out(capture)
 458
 459         # out2in 2nd interface
 460         pkts = self.create_stream_out(self.pg3)
 461         self.pg3.add_stream(pkts)
 462         self.pg_enable_capture(self.pg_interfaces)
 463         self.pg_start()
 464         capture = self.pg1.get_capture(len(pkts))
 465         self.verify_capture_in(capture, self.pg1)
 466
 467         # in2out 3rd interface
 468         pkts = self.create_stream_in(self.pg2, self.pg3)
 469         self.pg2.add_stream(pkts)
 470         self.pg_enable_capture(self.pg_interfaces)
 471         self.pg_start()
 472         capture = self.pg3.get_capture(len(pkts))
 473         self.verify_capture_out(capture)
 474
 475         # out2in 3rd interface
 476         pkts = self.create_stream_out(self.pg3)
 477         self.pg3.add_stream(pkts)
 478         self.pg_enable_capture(self.pg_interfaces)
 479         self.pg_start()
 480         capture = self.pg2.get_capture(len(pkts))
 481         self.verify_capture_in(capture, self.pg2)
 482
 483     def test_inside_overlapping_interfaces(self):
 484         """ SNAT multiple inside interfaces with overlapping address space """
 485
 486         self.snat_add_address(self.snat_addr)
 487         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 488                                                  is_inside=0)
 489         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 490         self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
 491         self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
 492
 493         # in2out 1st interface
 494         pkts = self.create_stream_in(self.pg4, self.pg3)
 495         self.pg4.add_stream(pkts)
 496         self.pg_enable_capture(self.pg_interfaces)
 497         self.pg_start()
 498         capture = self.pg3.get_capture(len(pkts))
 499         self.verify_capture_out(capture)
 500
 501         # out2in 1st interface
 502         pkts = self.create_stream_out(self.pg3)
 503         self.pg3.add_stream(pkts)
 504         self.pg_enable_capture(self.pg_interfaces)
 505         self.pg_start()
 506         capture = self.pg4.get_capture(len(pkts))
 507         self.verify_capture_in(capture, self.pg4)
 508
 509         # in2out 2nd interface
 510         pkts = self.create_stream_in(self.pg5, self.pg3)
 511         self.pg5.add_stream(pkts)
 512         self.pg_enable_capture(self.pg_interfaces)
 513         self.pg_start()
 514         capture = self.pg3.get_capture(len(pkts))
 515         self.verify_capture_out(capture)
 516
 517         # out2in 2nd interface
 518         pkts = self.create_stream_out(self.pg3)
 519         self.pg3.add_stream(pkts)
 520         self.pg_enable_capture(self.pg_interfaces)
 521         self.pg_start()
 522         capture = self.pg5.get_capture(len(pkts))
 523         self.verify_capture_in(capture, self.pg5)
 524
 525         # in2out 3rd interface
 526         pkts = self.create_stream_in(self.pg6, self.pg3)
 527         self.pg6.add_stream(pkts)
 528         self.pg_enable_capture(self.pg_interfaces)
 529         self.pg_start()
 530         capture = self.pg3.get_capture(len(pkts))
 531         self.verify_capture_out(capture)
 532
 533         # out2in 3rd interface
 534         pkts = self.create_stream_out(self.pg3)
 535         self.pg3.add_stream(pkts)
 536         self.pg_enable_capture(self.pg_interfaces)
 537         self.pg_start()
 538         capture = self.pg6.get_capture(len(pkts))
 539         self.verify_capture_in(capture, self.pg6)
 540
 541     def test_hairpinning(self):
 542         """ SNAT hairpinning """
 543
 544         host = self.pg0.remote_hosts[0]
 545         server = self.pg0.remote_hosts[1]
 546         host_in_port = 1234
 547         host_out_port = 0
 548         server_in_port = 5678
 549         server_out_port = 8765
 550
 551         self.snat_add_address(self.snat_addr)
 552         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 553         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 554                                                  is_inside=0)
 555         # add static mapping for server
 556         self.snat_add_static_mapping(server.ip4, self.snat_addr,
 557                                      server_in_port, server_out_port)
 558
 559         # send packet from host to server
 560         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
 561              IP(src=host.ip4, dst=self.snat_addr) /
 562              TCP(sport=host_in_port, dport=server_out_port))
 563         self.pg0.add_stream(p)
 564         self.pg_enable_capture(self.pg_interfaces)
 565         self.pg_start()
 566         capture = self.pg0.get_capture(1)
 567         p = capture[0]
 568         try:
 569             ip = p[IP]
 570             tcp = p[TCP]
 571             self.assertEqual(ip.src, self.snat_addr)
 572             self.assertEqual(ip.dst, server.ip4)
 573             self.assertNotEqual(tcp.sport, host_in_port)
 574             self.assertEqual(tcp.dport, server_in_port)
 575             host_out_port = tcp.sport
 576         except:
 577             self.logger.error(ppp("Unexpected or invalid packet:", p))
 578             raise
 579
 580         # send reply from server to host
 581         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
 582              IP(src=server.ip4, dst=self.snat_addr) /
 583              TCP(sport=server_in_port, dport=host_out_port))
 584         self.pg0.add_stream(p)
 585         self.pg_enable_capture(self.pg_interfaces)
 586         self.pg_start()
 587         capture = self.pg0.get_capture(1)
 588         p = capture[0]
 589         try:
 590             ip = p[IP]
 591             tcp = p[TCP]
 592             self.assertEqual(ip.src, self.snat_addr)
 593             self.assertEqual(ip.dst, host.ip4)
 594             self.assertEqual(tcp.sport, server_out_port)
 595             self.assertEqual(tcp.dport, host_in_port)
 596         except:
 597             self.logger.error(ppp("Unexpected or invalid packet:"), p)
 598             raise
 599
 600     def test_max_translations_per_user(self):
 601         """ MAX translations per user - recycle the least recently used """
 602
 603         self.snat_add_address(self.snat_addr)
 604         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 605         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 606                                                  is_inside=0)
 607
 608         # get maximum number of translations per user
 609         snat_config = self.vapi.snat_show_config()
 610
 611         # send more than maximum number of translations per user packets
 612         pkts_num = snat_config.max_translations_per_user + 5
 613         pkts = []
 614         for port in range(0, pkts_num):
 615             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 616                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 617                  TCP(sport=1025 + port))
 618             pkts.append(p)
 619         self.pg0.add_stream(pkts)
 620         self.pg_enable_capture(self.pg_interfaces)
 621         self.pg_start()
 622
 623         # verify number of translated packet
 624         self.pg1.get_capture(pkts_num)
 625
 626     def tearDown(self):
 627         super(TestSNAT, self).tearDown()
 628         if not self.vpp_dead:
 629             self.logger.info(self.vapi.cli("show snat verbose"))
 630             self.clear_snat()
 631
 632 if __name__ == '__main__':
 633     unittest.main(testRunner=VppTestRunner)