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(packet[TCP].sport, self.tcp_port_in)
 134                     self.tcp_port_out = packet[TCP].sport
 135                 elif packet.haslayer(UDP):
 136                     if same_port:
 137                         self.assertEqual(packet[UDP].sport, self.udp_port_in)
 138                     else:
 139                         self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
 140                     self.udp_port_out = packet[UDP].sport
 141                 else:
 142                     if same_port:
 143                         self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 144                     else:
 145                         self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
 146                     self.icmp_id_out = packet[ICMP].id
 147             except:
 148                 self.logger.error(ppp("Unexpected or invalid packet "
 149                                       "(outside network):", packet))
 150                 raise
 151
 152     def verify_capture_in(self, capture, in_if, packet_num=3):
 153         """
 154         Verify captured packets on inside network
 155
 156         :param capture: Captured packets
 157         :param in_if: Inside interface
 158         :param packet_num: Expected number of packets (Default 3)
 159         """
 160         self.assertEqual(packet_num, len(capture))
 161         for packet in capture:
 162             try:
 163                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 164                 if packet.haslayer(TCP):
 165                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 166                 elif packet.haslayer(UDP):
 167                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 168                 else:
 169                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 170             except:
 171                 self.logger.error(ppp("Unexpected or invalid packet "
 172                                       "(inside network):", packet))
 173                 raise
 174
 175     def clear_snat(self):
 176         """
 177         Clear SNAT configuration.
 178         """
 179         interfaces = self.vapi.snat_interface_dump()
 180         for intf in interfaces:
 181             self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
 182                                                      intf.is_inside,
 183                                                      is_add=0)
 184
 185         static_mappings = self.vapi.snat_static_mapping_dump()
 186         for sm in static_mappings:
 187             self.vapi.snat_add_static_mapping(sm.local_ip_address,
 188                                               sm.external_ip_address,
 189                                               local_port=sm.local_port,
 190                                               external_port=sm.external_port,
 191                                               addr_only=sm.addr_only,
 192                                               vrf_id=sm.vrf_id,
 193                                               is_add=0)
 194
 195         adresses = self.vapi.snat_address_dump()
 196         for addr in adresses:
 197             self.vapi.snat_add_address_range(addr.ip_address,
 198                                              addr.ip_address,
 199                                              is_add=0)
 200
 201     def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
 202                                 external_port=0, vrf_id=0, is_add=1):
 203         """
 204         Add/delete S-NAT static mapping
 205
 206         :param local_ip: Local IP address
 207         :param external_ip: External IP address
 208         :param local_port: Local port number (Optional)
 209         :param external_port: External port number (Optional)
 210         :param vrf_id: VRF ID (Default 0)
 211         :param is_add: 1 if add, 0 if delete (Default add)
 212         """
 213         addr_only = 1
 214         if local_port and external_port:
 215             addr_only = 0
 216         l_ip = socket.inet_pton(socket.AF_INET, local_ip)
 217         e_ip = socket.inet_pton(socket.AF_INET, external_ip)
 218         self.vapi.snat_add_static_mapping(l_ip, e_ip, local_port, external_port,
 219                                           addr_only, vrf_id, is_add)
 220
 221     def snat_add_address(self, ip, is_add=1):
 222         """
 223         Add/delete S-NAT address
 224
 225         :param ip: IP address
 226         :param is_add: 1 if add, 0 if delete (Default add)
 227         """
 228         snat_addr = socket.inet_pton(socket.AF_INET, ip)
 229         self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
 230
 231     def test_dynamic(self):
 232         """ SNAT dynamic translation test """
 233
 234         self.snat_add_address(self.snat_addr)
 235         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 236         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 237                                                  is_inside=0)
 238
 239         # in2out
 240         pkts = self.create_stream_in(self.pg0, self.pg1)
 241         self.pg0.add_stream(pkts)
 242         self.pg_enable_capture(self.pg_interfaces)
 243         self.pg_start()
 244         capture = self.pg1.get_capture(len(pkts))
 245         self.verify_capture_out(capture)
 246
 247         # out2in
 248         pkts = self.create_stream_out(self.pg1)
 249         self.pg1.add_stream(pkts)
 250         self.pg_enable_capture(self.pg_interfaces)
 251         self.pg_start()
 252         capture = self.pg0.get_capture(len(pkts))
 253         self.verify_capture_in(capture, self.pg0)
 254
 255     def test_static_in(self):
 256         """ SNAT 1:1 NAT initialized from inside network """
 257
 258         nat_ip = "10.0.0.10"
 259         self.tcp_port_out = 6303
 260         self.udp_port_out = 6304
 261         self.icmp_id_out = 6305
 262
 263         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 264         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 265         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 266                                                  is_inside=0)
 267
 268         # in2out
 269         pkts = self.create_stream_in(self.pg0, self.pg1)
 270         self.pg0.add_stream(pkts)
 271         self.pg_enable_capture(self.pg_interfaces)
 272         self.pg_start()
 273         capture = self.pg1.get_capture(len(pkts))
 274         self.verify_capture_out(capture, nat_ip, True)
 275
 276         # out2in
 277         pkts = self.create_stream_out(self.pg1, nat_ip)
 278         self.pg1.add_stream(pkts)
 279         self.pg_enable_capture(self.pg_interfaces)
 280         self.pg_start()
 281         capture = self.pg0.get_capture(len(pkts))
 282         self.verify_capture_in(capture, self.pg0)
 283
 284     def test_static_out(self):
 285         """ SNAT 1:1 NAT initialized from outside network """
 286
 287         nat_ip = "10.0.0.20"
 288         self.tcp_port_out = 6303
 289         self.udp_port_out = 6304
 290         self.icmp_id_out = 6305
 291
 292         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 293         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 294         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 295                                                  is_inside=0)
 296
 297         # out2in
 298         pkts = self.create_stream_out(self.pg1, nat_ip)
 299         self.pg1.add_stream(pkts)
 300         self.pg_enable_capture(self.pg_interfaces)
 301         self.pg_start()
 302         capture = self.pg0.get_capture(len(pkts))
 303         self.verify_capture_in(capture, self.pg0)
 304
 305         # in2out
 306         pkts = self.create_stream_in(self.pg0, self.pg1)
 307         self.pg0.add_stream(pkts)
 308         self.pg_enable_capture(self.pg_interfaces)
 309         self.pg_start()
 310         capture = self.pg1.get_capture(len(pkts))
 311         self.verify_capture_out(capture, nat_ip, True)
 312
 313     def test_static_with_port_in(self):
 314         """ SNAT 1:1 NAT with port initialized from inside network """
 315
 316         self.tcp_port_out = 3606
 317         self.udp_port_out = 3607
 318         self.icmp_id_out = 3608
 319
 320         self.snat_add_address(self.snat_addr)
 321         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 322                                      self.tcp_port_in, self.tcp_port_out)
 323         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 324                                      self.udp_port_in, self.udp_port_out)
 325         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 326                                      self.icmp_id_in, self.icmp_id_out)
 327         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 328         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 329                                                  is_inside=0)
 330
 331         # in2out
 332         pkts = self.create_stream_in(self.pg0, self.pg1)
 333         self.pg0.add_stream(pkts)
 334         self.pg_enable_capture(self.pg_interfaces)
 335         self.pg_start()
 336         capture = self.pg1.get_capture(len(pkts))
 337         self.verify_capture_out(capture)
 338
 339         # out2in
 340         pkts = self.create_stream_out(self.pg1)
 341         self.pg1.add_stream(pkts)
 342         self.pg_enable_capture(self.pg_interfaces)
 343         self.pg_start()
 344         capture = self.pg0.get_capture(len(pkts))
 345         self.verify_capture_in(capture, self.pg0)
 346
 347     def test_static_with_port_out(self):
 348         """ SNAT 1:1 NAT with port initialized from outside network """
 349
 350         self.tcp_port_out = 30606
 351         self.udp_port_out = 30607
 352         self.icmp_id_out = 30608
 353
 354         self.snat_add_address(self.snat_addr)
 355         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 356                                      self.tcp_port_in, self.tcp_port_out)
 357         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 358                                      self.udp_port_in, self.udp_port_out)
 359         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 360                                      self.icmp_id_in, self.icmp_id_out)
 361         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 362         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 363                                                  is_inside=0)
 364
 365         # out2in
 366         pkts = self.create_stream_out(self.pg1)
 367         self.pg1.add_stream(pkts)
 368         self.pg_enable_capture(self.pg_interfaces)
 369         self.pg_start()
 370         capture = self.pg0.get_capture(len(pkts))
 371         self.verify_capture_in(capture, self.pg0)
 372
 373         # in2out
 374         pkts = self.create_stream_in(self.pg0, self.pg1)
 375         self.pg0.add_stream(pkts)
 376         self.pg_enable_capture(self.pg_interfaces)
 377         self.pg_start()
 378         capture = self.pg1.get_capture(len(pkts))
 379         self.verify_capture_out(capture)
 380
 381     def test_static_vrf_aware(self):
 382         """ SNAT 1:1 NAT VRF awareness """
 383
 384         nat_ip1 = "10.0.0.30"
 385         nat_ip2 = "10.0.0.40"
 386         self.tcp_port_out = 6303
 387         self.udp_port_out = 6304
 388         self.icmp_id_out = 6305
 389
 390         self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
 391                                      vrf_id=self.pg4.sw_if_index)
 392         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
 393                                      vrf_id=self.pg4.sw_if_index)
 394         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 395                                                  is_inside=0)
 396         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 397         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 398
 399         # inside interface VRF match SNAT static mapping VRF
 400         pkts = self.create_stream_in(self.pg4, self.pg3)
 401         self.pg4.add_stream(pkts)
 402         self.pg_enable_capture(self.pg_interfaces)
 403         self.pg_start()
 404         capture = self.pg3.get_capture(len(pkts))
 405         self.verify_capture_out(capture, nat_ip1, True)
 406
 407         # inside interface VRF don't match SNAT static mapping VRF (packets
 408         # are dropped)
 409         pkts = self.create_stream_in(self.pg0, self.pg3)
 410         self.pg0.add_stream(pkts)
 411         self.pg_enable_capture(self.pg_interfaces)
 412         self.pg_start()
 413         self.pg3.assert_nothing_captured()
 414
 415     def test_multiple_inside_interfaces(self):
 416         """SNAT multiple inside interfaces with non-overlapping address space"""
 417
 418         self.snat_add_address(self.snat_addr)
 419         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 420         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
 421         self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index)
 422         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 423                                                  is_inside=0)
 424
 425         # in2out 1st interface
 426         pkts = self.create_stream_in(self.pg0, self.pg3)
 427         self.pg0.add_stream(pkts)
 428         self.pg_enable_capture(self.pg_interfaces)
 429         self.pg_start()
 430         capture = self.pg3.get_capture(len(pkts))
 431         self.verify_capture_out(capture)
 432
 433         # out2in 1st interface
 434         pkts = self.create_stream_out(self.pg3)
 435         self.pg3.add_stream(pkts)
 436         self.pg_enable_capture(self.pg_interfaces)
 437         self.pg_start()
 438         capture = self.pg0.get_capture(len(pkts))
 439         self.verify_capture_in(capture, self.pg0)
 440
 441         # in2out 2nd interface
 442         pkts = self.create_stream_in(self.pg1, self.pg3)
 443         self.pg1.add_stream(pkts)
 444         self.pg_enable_capture(self.pg_interfaces)
 445         self.pg_start()
 446         capture = self.pg3.get_capture(len(pkts))
 447         self.verify_capture_out(capture)
 448
 449         # out2in 2nd interface
 450         pkts = self.create_stream_out(self.pg3)
 451         self.pg3.add_stream(pkts)
 452         self.pg_enable_capture(self.pg_interfaces)
 453         self.pg_start()
 454         capture = self.pg1.get_capture(len(pkts))
 455         self.verify_capture_in(capture, self.pg1)
 456
 457         # in2out 3rd interface
 458         pkts = self.create_stream_in(self.pg2, self.pg3)
 459         self.pg2.add_stream(pkts)
 460         self.pg_enable_capture(self.pg_interfaces)
 461         self.pg_start()
 462         capture = self.pg3.get_capture(len(pkts))
 463         self.verify_capture_out(capture)
 464
 465         # out2in 3rd interface
 466         pkts = self.create_stream_out(self.pg3)
 467         self.pg3.add_stream(pkts)
 468         self.pg_enable_capture(self.pg_interfaces)
 469         self.pg_start()
 470         capture = self.pg2.get_capture(len(pkts))
 471         self.verify_capture_in(capture, self.pg2)
 472
 473     def test_inside_overlapping_interfaces(self):
 474         """ SNAT multiple inside interfaces with overlapping address space """
 475
 476         self.snat_add_address(self.snat_addr)
 477         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 478                                                  is_inside=0)
 479         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 480         self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
 481         self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
 482
 483         # in2out 1st interface
 484         pkts = self.create_stream_in(self.pg4, self.pg3)
 485         self.pg4.add_stream(pkts)
 486         self.pg_enable_capture(self.pg_interfaces)
 487         self.pg_start()
 488         capture = self.pg3.get_capture(len(pkts))
 489         self.verify_capture_out(capture)
 490
 491         # out2in 1st interface
 492         pkts = self.create_stream_out(self.pg3)
 493         self.pg3.add_stream(pkts)
 494         self.pg_enable_capture(self.pg_interfaces)
 495         self.pg_start()
 496         capture = self.pg4.get_capture(len(pkts))
 497         self.verify_capture_in(capture, self.pg4)
 498
 499         # in2out 2nd interface
 500         pkts = self.create_stream_in(self.pg5, self.pg3)
 501         self.pg5.add_stream(pkts)
 502         self.pg_enable_capture(self.pg_interfaces)
 503         self.pg_start()
 504         capture = self.pg3.get_capture(len(pkts))
 505         self.verify_capture_out(capture)
 506
 507         # out2in 2nd interface
 508         pkts = self.create_stream_out(self.pg3)
 509         self.pg3.add_stream(pkts)
 510         self.pg_enable_capture(self.pg_interfaces)
 511         self.pg_start()
 512         capture = self.pg5.get_capture(len(pkts))
 513         self.verify_capture_in(capture, self.pg5)
 514
 515         # in2out 3rd interface
 516         pkts = self.create_stream_in(self.pg6, self.pg3)
 517         self.pg6.add_stream(pkts)
 518         self.pg_enable_capture(self.pg_interfaces)
 519         self.pg_start()
 520         capture = self.pg3.get_capture(len(pkts))
 521         self.verify_capture_out(capture)
 522
 523         # out2in 3rd interface
 524         pkts = self.create_stream_out(self.pg3)
 525         self.pg3.add_stream(pkts)
 526         self.pg_enable_capture(self.pg_interfaces)
 527         self.pg_start()
 528         capture = self.pg6.get_capture(len(pkts))
 529         self.verify_capture_in(capture, self.pg6)
 530
 531     def test_hairpinning(self):
 532         """ SNAT hairpinning """
 533
 534         host = self.pg0.remote_hosts[0]
 535         server = self.pg0.remote_hosts[1]
 536         host_in_port = 1234
 537         host_out_port = 0
 538         server_in_port = 5678
 539         server_out_port = 8765
 540
 541         self.snat_add_address(self.snat_addr)
 542         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 543         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 544                                                  is_inside=0)
 545         # add static mapping for server
 546         self.snat_add_static_mapping(server.ip4, self.snat_addr,
 547                                      server_in_port, server_out_port)
 548
 549         # send packet from host to server
 550         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
 551              IP(src=host.ip4, dst=self.snat_addr) /
 552              TCP(sport=host_in_port, dport=server_out_port))
 553         self.pg0.add_stream(p)
 554         self.pg_enable_capture(self.pg_interfaces)
 555         self.pg_start()
 556         capture = self.pg0.get_capture(1)
 557         p = capture[0]
 558         try:
 559             ip = p[IP]
 560             tcp = p[TCP]
 561             self.assertEqual(ip.src, self.snat_addr)
 562             self.assertEqual(ip.dst, server.ip4)
 563             self.assertNotEqual(tcp.sport, host_in_port)
 564             self.assertEqual(tcp.dport, server_in_port)
 565             host_out_port = tcp.sport
 566         except:
 567             self.logger.error(ppp("Unexpected or invalid packet:", p))
 568             raise
 569
 570         # send reply from server to host
 571         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
 572              IP(src=server.ip4, dst=self.snat_addr) /
 573              TCP(sport=server_in_port, dport=host_out_port))
 574         self.pg0.add_stream(p)
 575         self.pg_enable_capture(self.pg_interfaces)
 576         self.pg_start()
 577         capture = self.pg0.get_capture(1)
 578         p = capture[0]
 579         try:
 580             ip = p[IP]
 581             tcp = p[TCP]
 582             self.assertEqual(ip.src, self.snat_addr)
 583             self.assertEqual(ip.dst, host.ip4)
 584             self.assertEqual(tcp.sport, server_out_port)
 585             self.assertEqual(tcp.dport, host_in_port)
 586         except:
 587             self.logger.error(ppp("Unexpected or invalid packet:"), p)
 588             raise
 589
 590     def test_max_translations_per_user(self):
 591         """ MAX translations per user - recycle the least recently used """
 592
 593         self.snat_add_address(self.snat_addr)
 594         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 595         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 596                                                  is_inside=0)
 597
 598         # get maximum number of translations per user
 599         snat_config = self.vapi.snat_show_config()
 600
 601         # send more than maximum number of translations per user packets
 602         pkts_num = snat_config.max_translations_per_user + 5
 603         pkts = []
 604         for port in range(0, pkts_num):
 605             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 606                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 607                  TCP(sport=1025 + port))
 608             pkts.append(p)
 609         self.pg0.add_stream(pkts)
 610         self.pg_enable_capture(self.pg_interfaces)
 611         self.pg_start()
 612
 613         # verify number of translated packet
 614         self.pg1.get_capture(pkts_num)
 615
 616     def tearDown(self):
 617         super(TestSNAT, self).tearDown()
 618         if not self.vpp_dead:
 619             self.logger.info(self.vapi.cli("show snat verbose"))
 620             self.clear_snat()
 621
 622 if __name__ == '__main__':
 623     unittest.main(testRunner=VppTestRunner)