test/test_snat.py

   1 #!/usr/bin/env python
   2
   3 import socket
   4 import unittest
   5 import struct
   6
   7 from framework import VppTestCase, VppTestRunner
   8 from scapy.layers.inet import IP, TCP, UDP, ICMP
   9 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
  10 from scapy.layers.l2 import Ether, ARP
  11 from scapy.data import IP_PROTOS
  12 from util import ppp
  13 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
  14
  15
  16 class TestSNAT(VppTestCase):
  17     """ SNAT Test Cases """
  18
  19     @classmethod
  20     def setUpClass(cls):
  21         super(TestSNAT, cls).setUpClass()
  22
  23         try:
  24             cls.tcp_port_in = 6303
  25             cls.tcp_port_out = 6303
  26             cls.udp_port_in = 6304
  27             cls.udp_port_out = 6304
  28             cls.icmp_id_in = 6305
  29             cls.icmp_id_out = 6305
  30             cls.snat_addr = '10.0.0.3'
  31
  32             cls.create_pg_interfaces(range(8))
  33             cls.interfaces = list(cls.pg_interfaces[0:4])
  34
  35             for i in cls.interfaces:
  36                 i.admin_up()
  37                 i.config_ip4()
  38                 i.resolve_arp()
  39
  40             cls.pg0.generate_remote_hosts(2)
  41             cls.pg0.configure_ipv4_neighbors()
  42
  43             cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
  44
  45             cls.pg4._local_ip4 = "172.16.255.1"
  46             cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
  47             cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
  48             cls.pg4.set_table_ip4(10)
  49             cls.pg5._local_ip4 = "172.16.255.3"
  50             cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
  51             cls.pg5._remote_hosts[0]._ip4 = "172.16.255.4"
  52             cls.pg5.set_table_ip4(10)
  53             cls.pg6._local_ip4 = "172.16.255.1"
  54             cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
  55             cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
  56             cls.pg6.set_table_ip4(20)
  57             for i in cls.overlapping_interfaces:
  58                 i.config_ip4()
  59                 i.admin_up()
  60                 i.resolve_arp()
  61
  62             cls.pg7.admin_up()
  63
  64         except Exception:
  65             super(TestSNAT, cls).tearDownClass()
  66             raise
  67
  68     def create_stream_in(self, in_if, out_if, ttl=64):
  69         """
  70         Create packet stream for inside network
  71
  72         :param in_if: Inside interface
  73         :param out_if: Outside interface
  74         :param ttl: TTL of generated packets
  75         """
  76         pkts = []
  77         # TCP
  78         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  79              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
  80              TCP(sport=self.tcp_port_in))
  81         pkts.append(p)
  82
  83         # UDP
  84         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  85              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
  86              UDP(sport=self.udp_port_in))
  87         pkts.append(p)
  88
  89         # ICMP
  90         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
  91              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
  92              ICMP(id=self.icmp_id_in, type='echo-request'))
  93         pkts.append(p)
  94
  95         return pkts
  96
  97     def create_stream_out(self, out_if, dst_ip=None, ttl=64):
  98         """
  99         Create packet stream for outside network
 100
 101         :param out_if: Outside interface
 102         :param dst_ip: Destination IP address (Default use global SNAT address)
 103         :param ttl: TTL of generated packets
 104         """
 105         if dst_ip is None:
 106             dst_ip = self.snat_addr
 107         pkts = []
 108         # TCP
 109         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 110              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 111              TCP(dport=self.tcp_port_out))
 112         pkts.append(p)
 113
 114         # UDP
 115         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 116              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 117              UDP(dport=self.udp_port_out))
 118         pkts.append(p)
 119
 120         # ICMP
 121         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 122              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 123              ICMP(id=self.icmp_id_out, type='echo-reply'))
 124         pkts.append(p)
 125
 126         return pkts
 127
 128     def verify_capture_out(self, capture, nat_ip=None, same_port=False,
 129                            packet_num=3):
 130         """
 131         Verify captured packets on outside network
 132
 133         :param capture: Captured packets
 134         :param nat_ip: Translated IP address (Default use global SNAT address)
 135         :param same_port: Sorce port number is not translated (Default False)
 136         :param packet_num: Expected number of packets (Default 3)
 137         """
 138         if nat_ip is None:
 139             nat_ip = self.snat_addr
 140         self.assertEqual(packet_num, len(capture))
 141         for packet in capture:
 142             try:
 143                 self.assertEqual(packet[IP].src, nat_ip)
 144                 if packet.haslayer(TCP):
 145                     if same_port:
 146                         self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 147                     else:
 148                         self.assertNotEqual(
 149                             packet[TCP].sport, self.tcp_port_in)
 150                     self.tcp_port_out = packet[TCP].sport
 151                 elif packet.haslayer(UDP):
 152                     if same_port:
 153                         self.assertEqual(packet[UDP].sport, self.udp_port_in)
 154                     else:
 155                         self.assertNotEqual(
 156                             packet[UDP].sport, self.udp_port_in)
 157                     self.udp_port_out = packet[UDP].sport
 158                 else:
 159                     if same_port:
 160                         self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 161                     else:
 162                         self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
 163                     self.icmp_id_out = packet[ICMP].id
 164             except:
 165                 self.logger.error(ppp("Unexpected or invalid packet "
 166                                       "(outside network):", packet))
 167                 raise
 168
 169     def verify_capture_in(self, capture, in_if, packet_num=3):
 170         """
 171         Verify captured packets on inside network
 172
 173         :param capture: Captured packets
 174         :param in_if: Inside interface
 175         :param packet_num: Expected number of packets (Default 3)
 176         """
 177         self.assertEqual(packet_num, len(capture))
 178         for packet in capture:
 179             try:
 180                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 181                 if packet.haslayer(TCP):
 182                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 183                 elif packet.haslayer(UDP):
 184                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 185                 else:
 186                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 187             except:
 188                 self.logger.error(ppp("Unexpected or invalid packet "
 189                                       "(inside network):", packet))
 190                 raise
 191
 192     def verify_capture_no_translation(self, capture, ingress_if, egress_if):
 193         """
 194         Verify captured packet that don't have to be translated
 195
 196         :param capture: Captured packets
 197         :param ingress_if: Ingress interface
 198         :param egress_if: Egress interface
 199         """
 200         for packet in capture:
 201             try:
 202                 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
 203                 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
 204                 if packet.haslayer(TCP):
 205                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 206                 elif packet.haslayer(UDP):
 207                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
 208                 else:
 209                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 210             except:
 211                 self.logger.error(ppp("Unexpected or invalid packet "
 212                                       "(inside network):", packet))
 213                 raise
 214
 215     def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
 216                                             packet_num=3, icmp_type=11):
 217         """
 218         Verify captured packets with ICMP errors on outside network
 219
 220         :param capture: Captured packets
 221         :param src_ip: Translated IP address or IP address of VPP
 222                        (Default use global SNAT address)
 223         :param packet_num: Expected number of packets (Default 3)
 224         :param icmp_type: Type of error ICMP packet
 225                           we are expecting (Default 11)
 226         """
 227         if src_ip is None:
 228             src_ip = self.snat_addr
 229         self.assertEqual(packet_num, len(capture))
 230         for packet in capture:
 231             try:
 232                 self.assertEqual(packet[IP].src, src_ip)
 233                 self.assertTrue(packet.haslayer(ICMP))
 234                 icmp = packet[ICMP]
 235                 self.assertEqual(icmp.type, icmp_type)
 236                 self.assertTrue(icmp.haslayer(IPerror))
 237                 inner_ip = icmp[IPerror]
 238                 if inner_ip.haslayer(TCPerror):
 239                     self.assertEqual(inner_ip[TCPerror].dport,
 240                                      self.tcp_port_out)
 241                 elif inner_ip.haslayer(UDPerror):
 242                     self.assertEqual(inner_ip[UDPerror].dport,
 243                                      self.udp_port_out)
 244                 else:
 245                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
 246             except:
 247                 self.logger.error(ppp("Unexpected or invalid packet "
 248                                       "(outside network):", packet))
 249                 raise
 250
 251     def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
 252                                            icmp_type=11):
 253         """
 254         Verify captured packets with ICMP errors on inside network
 255
 256         :param capture: Captured packets
 257         :param in_if: Inside interface
 258         :param packet_num: Expected number of packets (Default 3)
 259         :param icmp_type: Type of error ICMP packet
 260                           we are expecting (Default 11)
 261         """
 262         self.assertEqual(packet_num, len(capture))
 263         for packet in capture:
 264             try:
 265                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 266                 self.assertTrue(packet.haslayer(ICMP))
 267                 icmp = packet[ICMP]
 268                 self.assertEqual(icmp.type, icmp_type)
 269                 self.assertTrue(icmp.haslayer(IPerror))
 270                 inner_ip = icmp[IPerror]
 271                 if inner_ip.haslayer(TCPerror):
 272                     self.assertEqual(inner_ip[TCPerror].sport,
 273                                      self.tcp_port_in)
 274                 elif inner_ip.haslayer(UDPerror):
 275                     self.assertEqual(inner_ip[UDPerror].sport,
 276                                      self.udp_port_in)
 277                 else:
 278                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
 279             except:
 280                 self.logger.error(ppp("Unexpected or invalid packet "
 281                                       "(inside network):", packet))
 282                 raise
 283
 284     def verify_ipfix_nat44_ses(self, data):
 285         """
 286         Verify IPFIX NAT44 session create/delete event
 287
 288         :param data: Decoded IPFIX data records
 289         """
 290         nat44_ses_create_num = 0
 291         nat44_ses_delete_num = 0
 292         self.assertEqual(6, len(data))
 293         for record in data:
 294             # natEvent
 295             self.assertIn(ord(record[230]), [4, 5])
 296             if ord(record[230]) == 4:
 297                 nat44_ses_create_num += 1
 298             else:
 299                 nat44_ses_delete_num += 1
 300             # sourceIPv4Address
 301             self.assertEqual(self.pg0.remote_ip4n, record[8])
 302             # postNATSourceIPv4Address
 303             self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
 304                              record[225])
 305             # ingressVRFID
 306             self.assertEqual(struct.pack("!I", 0), record[234])
 307             # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
 308             if IP_PROTOS.icmp == ord(record[4]):
 309                 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
 310                 self.assertEqual(struct.pack("!H", self.icmp_id_out),
 311                                  record[227])
 312             elif IP_PROTOS.tcp == ord(record[4]):
 313                 self.assertEqual(struct.pack("!H", self.tcp_port_in),
 314                                  record[7])
 315                 self.assertEqual(struct.pack("!H", self.tcp_port_out),
 316                                  record[227])
 317             elif IP_PROTOS.udp == ord(record[4]):
 318                 self.assertEqual(struct.pack("!H", self.udp_port_in),
 319                                  record[7])
 320                 self.assertEqual(struct.pack("!H", self.udp_port_out),
 321                                  record[227])
 322             else:
 323                 self.fail("Invalid protocol")
 324         self.assertEqual(3, nat44_ses_create_num)
 325         self.assertEqual(3, nat44_ses_delete_num)
 326
 327     def verify_ipfix_addr_exhausted(self, data):
 328         """
 329         Verify IPFIX NAT addresses event
 330
 331         :param data: Decoded IPFIX data records
 332         """
 333         self.assertEqual(1, len(data))
 334         record = data[0]
 335         # natEvent
 336         self.assertEqual(ord(record[230]), 3)
 337         # natPoolID
 338         self.assertEqual(struct.pack("!I", 0), record[283])
 339
 340     def clear_snat(self):
 341         """
 342         Clear SNAT configuration.
 343         """
 344         if self.pg7.has_ip4_config:
 345             self.pg7.unconfig_ip4()
 346
 347         interfaces = self.vapi.snat_interface_addr_dump()
 348         for intf in interfaces:
 349             self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
 350
 351         self.vapi.snat_ipfix(enable=0)
 352
 353         interfaces = self.vapi.snat_interface_dump()
 354         for intf in interfaces:
 355             self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
 356                                                      intf.is_inside,
 357                                                      is_add=0)
 358
 359         static_mappings = self.vapi.snat_static_mapping_dump()
 360         for sm in static_mappings:
 361             self.vapi.snat_add_static_mapping(sm.local_ip_address,
 362                                               sm.external_ip_address,
 363                                               local_port=sm.local_port,
 364                                               external_port=sm.external_port,
 365                                               addr_only=sm.addr_only,
 366                                               vrf_id=sm.vrf_id,
 367                                               protocol=sm.protocol,
 368                                               is_add=0)
 369
 370         adresses = self.vapi.snat_address_dump()
 371         for addr in adresses:
 372             self.vapi.snat_add_address_range(addr.ip_address,
 373                                              addr.ip_address,
 374                                              is_add=0)
 375
 376     def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
 377                                 local_port=0, external_port=0, vrf_id=0,
 378                                 is_add=1, external_sw_if_index=0xFFFFFFFF,
 379                                 proto=0):
 380         """
 381         Add/delete S-NAT static mapping
 382
 383         :param local_ip: Local IP address
 384         :param external_ip: External IP address
 385         :param local_port: Local port number (Optional)
 386         :param external_port: External port number (Optional)
 387         :param vrf_id: VRF ID (Default 0)
 388         :param is_add: 1 if add, 0 if delete (Default add)
 389         :param external_sw_if_index: External interface instead of IP address
 390         :param proto: IP protocol (Mandatory if port specified)
 391         """
 392         addr_only = 1
 393         if local_port and external_port:
 394             addr_only = 0
 395         l_ip = socket.inet_pton(socket.AF_INET, local_ip)
 396         e_ip = socket.inet_pton(socket.AF_INET, external_ip)
 397         self.vapi.snat_add_static_mapping(
 398             l_ip,
 399             e_ip,
 400             external_sw_if_index,
 401             local_port,
 402             external_port,
 403             addr_only,
 404             vrf_id,
 405             proto,
 406             is_add)
 407
 408     def snat_add_address(self, ip, is_add=1):
 409         """
 410         Add/delete S-NAT address
 411
 412         :param ip: IP address
 413         :param is_add: 1 if add, 0 if delete (Default add)
 414         """
 415         snat_addr = socket.inet_pton(socket.AF_INET, ip)
 416         self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add)
 417
 418     def test_dynamic(self):
 419         """ SNAT dynamic translation test """
 420
 421         self.snat_add_address(self.snat_addr)
 422         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 423         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 424                                                  is_inside=0)
 425
 426         # in2out
 427         pkts = self.create_stream_in(self.pg0, self.pg1)
 428         self.pg0.add_stream(pkts)
 429         self.pg_enable_capture(self.pg_interfaces)
 430         self.pg_start()
 431         capture = self.pg1.get_capture(len(pkts))
 432         self.verify_capture_out(capture)
 433
 434         # out2in
 435         pkts = self.create_stream_out(self.pg1)
 436         self.pg1.add_stream(pkts)
 437         self.pg_enable_capture(self.pg_interfaces)
 438         self.pg_start()
 439         capture = self.pg0.get_capture(len(pkts))
 440         self.verify_capture_in(capture, self.pg0)
 441
 442     def test_dynamic_icmp_errors_in2out_ttl_1(self):
 443         """ SNAT handling of client packets with TTL=1 """
 444
 445         self.snat_add_address(self.snat_addr)
 446         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 447         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 448                                                  is_inside=0)
 449
 450         # Client side - generate traffic
 451         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
 452         self.pg0.add_stream(pkts)
 453         self.pg_enable_capture(self.pg_interfaces)
 454         self.pg_start()
 455
 456         # Client side - verify ICMP type 11 packets
 457         capture = self.pg0.get_capture(len(pkts))
 458         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
 459
 460     def test_dynamic_icmp_errors_out2in_ttl_1(self):
 461         """ SNAT handling of server packets with TTL=1 """
 462
 463         self.snat_add_address(self.snat_addr)
 464         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 465         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 466                                                  is_inside=0)
 467
 468         # Client side - create sessions
 469         pkts = self.create_stream_in(self.pg0, self.pg1)
 470         self.pg0.add_stream(pkts)
 471         self.pg_enable_capture(self.pg_interfaces)
 472         self.pg_start()
 473
 474         # Server side - generate traffic
 475         capture = self.pg1.get_capture(len(pkts))
 476         self.verify_capture_out(capture)
 477         pkts = self.create_stream_out(self.pg1, ttl=1)
 478         self.pg1.add_stream(pkts)
 479         self.pg_enable_capture(self.pg_interfaces)
 480         self.pg_start()
 481
 482         # Server side - verify ICMP type 11 packets
 483         capture = self.pg1.get_capture(len(pkts))
 484         self.verify_capture_out_with_icmp_errors(capture,
 485                                                  src_ip=self.pg1.local_ip4)
 486
 487     def test_dynamic_icmp_errors_in2out_ttl_2(self):
 488         """ SNAT handling of error responses to client packets with TTL=2 """
 489
 490         self.snat_add_address(self.snat_addr)
 491         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 492         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 493                                                  is_inside=0)
 494
 495         # Client side - generate traffic
 496         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
 497         self.pg0.add_stream(pkts)
 498         self.pg_enable_capture(self.pg_interfaces)
 499         self.pg_start()
 500
 501         # Server side - simulate ICMP type 11 response
 502         capture = self.pg1.get_capture(len(pkts))
 503         pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 504                 IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
 505                 ICMP(type=11) / packet[IP] for packet in capture]
 506         self.pg1.add_stream(pkts)
 507         self.pg_enable_capture(self.pg_interfaces)
 508         self.pg_start()
 509
 510         # Client side - verify ICMP type 11 packets
 511         capture = self.pg0.get_capture(len(pkts))
 512         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
 513
 514     def test_dynamic_icmp_errors_out2in_ttl_2(self):
 515         """ SNAT handling of error responses to server packets with TTL=2 """
 516
 517         self.snat_add_address(self.snat_addr)
 518         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 519         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 520                                                  is_inside=0)
 521
 522         # Client side - create sessions
 523         pkts = self.create_stream_in(self.pg0, self.pg1)
 524         self.pg0.add_stream(pkts)
 525         self.pg_enable_capture(self.pg_interfaces)
 526         self.pg_start()
 527
 528         # Server side - generate traffic
 529         capture = self.pg1.get_capture(len(pkts))
 530         self.verify_capture_out(capture)
 531         pkts = self.create_stream_out(self.pg1, ttl=2)
 532         self.pg1.add_stream(pkts)
 533         self.pg_enable_capture(self.pg_interfaces)
 534         self.pg_start()
 535
 536         # Client side - simulate ICMP type 11 response
 537         capture = self.pg0.get_capture(len(pkts))
 538         pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 539                 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 540                 ICMP(type=11) / packet[IP] for packet in capture]
 541         self.pg0.add_stream(pkts)
 542         self.pg_enable_capture(self.pg_interfaces)
 543         self.pg_start()
 544
 545         # Server side - verify ICMP type 11 packets
 546         capture = self.pg1.get_capture(len(pkts))
 547         self.verify_capture_out_with_icmp_errors(capture)
 548
 549     def test_ping_out_interface_from_outside(self):
 550         """ Ping SNAT out interface from outside """
 551
 552         self.snat_add_address(self.snat_addr)
 553         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 554         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 555                                                  is_inside=0)
 556
 557         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
 558              IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
 559              ICMP(id=self.icmp_id_out, type='echo-request'))
 560         pkts = [p]
 561         self.pg1.add_stream(pkts)
 562         self.pg_enable_capture(self.pg_interfaces)
 563         self.pg_start()
 564         capture = self.pg1.get_capture(len(pkts))
 565         self.assertEqual(1, len(capture))
 566         packet = capture[0]
 567         try:
 568             self.assertEqual(packet[IP].src, self.pg1.local_ip4)
 569             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
 570             self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 571             self.assertEqual(packet[ICMP].type, 0)  # echo reply
 572         except:
 573             self.logger.error(ppp("Unexpected or invalid packet "
 574                                   "(outside network):", packet))
 575             raise
 576
 577     def test_static_in(self):
 578         """ SNAT 1:1 NAT initialized from inside network """
 579
 580         nat_ip = "10.0.0.10"
 581         self.tcp_port_out = 6303
 582         self.udp_port_out = 6304
 583         self.icmp_id_out = 6305
 584
 585         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 586         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 587         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 588                                                  is_inside=0)
 589
 590         # in2out
 591         pkts = self.create_stream_in(self.pg0, self.pg1)
 592         self.pg0.add_stream(pkts)
 593         self.pg_enable_capture(self.pg_interfaces)
 594         self.pg_start()
 595         capture = self.pg1.get_capture(len(pkts))
 596         self.verify_capture_out(capture, nat_ip, True)
 597
 598         # out2in
 599         pkts = self.create_stream_out(self.pg1, nat_ip)
 600         self.pg1.add_stream(pkts)
 601         self.pg_enable_capture(self.pg_interfaces)
 602         self.pg_start()
 603         capture = self.pg0.get_capture(len(pkts))
 604         self.verify_capture_in(capture, self.pg0)
 605
 606     def test_static_out(self):
 607         """ SNAT 1:1 NAT initialized from outside network """
 608
 609         nat_ip = "10.0.0.20"
 610         self.tcp_port_out = 6303
 611         self.udp_port_out = 6304
 612         self.icmp_id_out = 6305
 613
 614         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
 615         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 616         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 617                                                  is_inside=0)
 618
 619         # out2in
 620         pkts = self.create_stream_out(self.pg1, nat_ip)
 621         self.pg1.add_stream(pkts)
 622         self.pg_enable_capture(self.pg_interfaces)
 623         self.pg_start()
 624         capture = self.pg0.get_capture(len(pkts))
 625         self.verify_capture_in(capture, self.pg0)
 626
 627         # in2out
 628         pkts = self.create_stream_in(self.pg0, self.pg1)
 629         self.pg0.add_stream(pkts)
 630         self.pg_enable_capture(self.pg_interfaces)
 631         self.pg_start()
 632         capture = self.pg1.get_capture(len(pkts))
 633         self.verify_capture_out(capture, nat_ip, True)
 634
 635     def test_static_with_port_in(self):
 636         """ SNAT 1:1 NAT with port initialized from inside network """
 637
 638         self.tcp_port_out = 3606
 639         self.udp_port_out = 3607
 640         self.icmp_id_out = 3608
 641
 642         self.snat_add_address(self.snat_addr)
 643         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 644                                      self.tcp_port_in, self.tcp_port_out,
 645                                      proto=IP_PROTOS.tcp)
 646         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 647                                      self.udp_port_in, self.udp_port_out,
 648                                      proto=IP_PROTOS.udp)
 649         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 650                                      self.icmp_id_in, self.icmp_id_out,
 651                                      proto=IP_PROTOS.icmp)
 652         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 653         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 654                                                  is_inside=0)
 655
 656         # in2out
 657         pkts = self.create_stream_in(self.pg0, self.pg1)
 658         self.pg0.add_stream(pkts)
 659         self.pg_enable_capture(self.pg_interfaces)
 660         self.pg_start()
 661         capture = self.pg1.get_capture(len(pkts))
 662         self.verify_capture_out(capture)
 663
 664         # out2in
 665         pkts = self.create_stream_out(self.pg1)
 666         self.pg1.add_stream(pkts)
 667         self.pg_enable_capture(self.pg_interfaces)
 668         self.pg_start()
 669         capture = self.pg0.get_capture(len(pkts))
 670         self.verify_capture_in(capture, self.pg0)
 671
 672     def test_static_with_port_out(self):
 673         """ SNAT 1:1 NAT with port initialized from outside network """
 674
 675         self.tcp_port_out = 30606
 676         self.udp_port_out = 30607
 677         self.icmp_id_out = 30608
 678
 679         self.snat_add_address(self.snat_addr)
 680         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 681                                      self.tcp_port_in, self.tcp_port_out,
 682                                      proto=IP_PROTOS.tcp)
 683         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 684                                      self.udp_port_in, self.udp_port_out,
 685                                      proto=IP_PROTOS.udp)
 686         self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
 687                                      self.icmp_id_in, self.icmp_id_out,
 688                                      proto=IP_PROTOS.icmp)
 689         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 690         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 691                                                  is_inside=0)
 692
 693         # out2in
 694         pkts = self.create_stream_out(self.pg1)
 695         self.pg1.add_stream(pkts)
 696         self.pg_enable_capture(self.pg_interfaces)
 697         self.pg_start()
 698         capture = self.pg0.get_capture(len(pkts))
 699         self.verify_capture_in(capture, self.pg0)
 700
 701         # in2out
 702         pkts = self.create_stream_in(self.pg0, self.pg1)
 703         self.pg0.add_stream(pkts)
 704         self.pg_enable_capture(self.pg_interfaces)
 705         self.pg_start()
 706         capture = self.pg1.get_capture(len(pkts))
 707         self.verify_capture_out(capture)
 708
 709     def test_static_vrf_aware(self):
 710         """ SNAT 1:1 NAT VRF awareness """
 711
 712         nat_ip1 = "10.0.0.30"
 713         nat_ip2 = "10.0.0.40"
 714         self.tcp_port_out = 6303
 715         self.udp_port_out = 6304
 716         self.icmp_id_out = 6305
 717
 718         self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
 719                                      vrf_id=10)
 720         self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
 721                                      vrf_id=10)
 722         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 723                                                  is_inside=0)
 724         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 725         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 726
 727         # inside interface VRF match SNAT static mapping VRF
 728         pkts = self.create_stream_in(self.pg4, self.pg3)
 729         self.pg4.add_stream(pkts)
 730         self.pg_enable_capture(self.pg_interfaces)
 731         self.pg_start()
 732         capture = self.pg3.get_capture(len(pkts))
 733         self.verify_capture_out(capture, nat_ip1, True)
 734
 735         # inside interface VRF don't match SNAT static mapping VRF (packets
 736         # are dropped)
 737         pkts = self.create_stream_in(self.pg0, self.pg3)
 738         self.pg0.add_stream(pkts)
 739         self.pg_enable_capture(self.pg_interfaces)
 740         self.pg_start()
 741         self.pg3.assert_nothing_captured()
 742
 743     def test_multiple_inside_interfaces(self):
 744         """ SNAT multiple inside interfaces (non-overlapping address space) """
 745
 746         self.snat_add_address(self.snat_addr)
 747         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 748         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
 749         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 750                                                  is_inside=0)
 751
 752         # between two S-NAT inside interfaces (no translation)
 753         pkts = self.create_stream_in(self.pg0, self.pg1)
 754         self.pg0.add_stream(pkts)
 755         self.pg_enable_capture(self.pg_interfaces)
 756         self.pg_start()
 757         capture = self.pg1.get_capture(len(pkts))
 758         self.verify_capture_no_translation(capture, self.pg0, self.pg1)
 759
 760         # from S-NAT inside to interface without S-NAT feature (no translation)
 761         pkts = self.create_stream_in(self.pg0, self.pg2)
 762         self.pg0.add_stream(pkts)
 763         self.pg_enable_capture(self.pg_interfaces)
 764         self.pg_start()
 765         capture = self.pg2.get_capture(len(pkts))
 766         self.verify_capture_no_translation(capture, self.pg0, self.pg2)
 767
 768         # in2out 1st interface
 769         pkts = self.create_stream_in(self.pg0, self.pg3)
 770         self.pg0.add_stream(pkts)
 771         self.pg_enable_capture(self.pg_interfaces)
 772         self.pg_start()
 773         capture = self.pg3.get_capture(len(pkts))
 774         self.verify_capture_out(capture)
 775
 776         # out2in 1st interface
 777         pkts = self.create_stream_out(self.pg3)
 778         self.pg3.add_stream(pkts)
 779         self.pg_enable_capture(self.pg_interfaces)
 780         self.pg_start()
 781         capture = self.pg0.get_capture(len(pkts))
 782         self.verify_capture_in(capture, self.pg0)
 783
 784         # in2out 2nd interface
 785         pkts = self.create_stream_in(self.pg1, self.pg3)
 786         self.pg1.add_stream(pkts)
 787         self.pg_enable_capture(self.pg_interfaces)
 788         self.pg_start()
 789         capture = self.pg3.get_capture(len(pkts))
 790         self.verify_capture_out(capture)
 791
 792         # out2in 2nd interface
 793         pkts = self.create_stream_out(self.pg3)
 794         self.pg3.add_stream(pkts)
 795         self.pg_enable_capture(self.pg_interfaces)
 796         self.pg_start()
 797         capture = self.pg1.get_capture(len(pkts))
 798         self.verify_capture_in(capture, self.pg1)
 799
 800     def test_inside_overlapping_interfaces(self):
 801         """ SNAT multiple inside interfaces with overlapping address space """
 802
 803         static_nat_ip = "10.0.0.10"
 804         self.snat_add_address(self.snat_addr)
 805         self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
 806                                                  is_inside=0)
 807         self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
 808         self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
 809         self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
 810         self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
 811                                      vrf_id=20)
 812
 813         # between S-NAT inside interfaces with same VRF (no translation)
 814         pkts = self.create_stream_in(self.pg4, self.pg5)
 815         self.pg4.add_stream(pkts)
 816         self.pg_enable_capture(self.pg_interfaces)
 817         self.pg_start()
 818         capture = self.pg5.get_capture(len(pkts))
 819         self.verify_capture_no_translation(capture, self.pg4, self.pg5)
 820
 821         # between S-NAT inside interfaces with different VRF (hairpinning)
 822         p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
 823              IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
 824              TCP(sport=1234, dport=5678))
 825         self.pg4.add_stream(p)
 826         self.pg_enable_capture(self.pg_interfaces)
 827         self.pg_start()
 828         capture = self.pg6.get_capture(1)
 829         p = capture[0]
 830         try:
 831             ip = p[IP]
 832             tcp = p[TCP]
 833             self.assertEqual(ip.src, self.snat_addr)
 834             self.assertEqual(ip.dst, self.pg6.remote_ip4)
 835             self.assertNotEqual(tcp.sport, 1234)
 836             self.assertEqual(tcp.dport, 5678)
 837         except:
 838             self.logger.error(ppp("Unexpected or invalid packet:", p))
 839             raise
 840
 841         # in2out 1st interface
 842         pkts = self.create_stream_in(self.pg4, self.pg3)
 843         self.pg4.add_stream(pkts)
 844         self.pg_enable_capture(self.pg_interfaces)
 845         self.pg_start()
 846         capture = self.pg3.get_capture(len(pkts))
 847         self.verify_capture_out(capture)
 848
 849         # out2in 1st interface
 850         pkts = self.create_stream_out(self.pg3)
 851         self.pg3.add_stream(pkts)
 852         self.pg_enable_capture(self.pg_interfaces)
 853         self.pg_start()
 854         capture = self.pg4.get_capture(len(pkts))
 855         self.verify_capture_in(capture, self.pg4)
 856
 857         # in2out 2nd interface
 858         pkts = self.create_stream_in(self.pg5, self.pg3)
 859         self.pg5.add_stream(pkts)
 860         self.pg_enable_capture(self.pg_interfaces)
 861         self.pg_start()
 862         capture = self.pg3.get_capture(len(pkts))
 863         self.verify_capture_out(capture)
 864
 865         # out2in 2nd interface
 866         pkts = self.create_stream_out(self.pg3)
 867         self.pg3.add_stream(pkts)
 868         self.pg_enable_capture(self.pg_interfaces)
 869         self.pg_start()
 870         capture = self.pg5.get_capture(len(pkts))
 871         self.verify_capture_in(capture, self.pg5)
 872
 873         # in2out 3rd interface
 874         pkts = self.create_stream_in(self.pg6, self.pg3)
 875         self.pg6.add_stream(pkts)
 876         self.pg_enable_capture(self.pg_interfaces)
 877         self.pg_start()
 878         capture = self.pg3.get_capture(len(pkts))
 879         self.verify_capture_out(capture, static_nat_ip, True)
 880
 881         # out2in 3rd interface
 882         pkts = self.create_stream_out(self.pg3, static_nat_ip)
 883         self.pg3.add_stream(pkts)
 884         self.pg_enable_capture(self.pg_interfaces)
 885         self.pg_start()
 886         capture = self.pg6.get_capture(len(pkts))
 887         self.verify_capture_in(capture, self.pg6)
 888
 889     def test_hairpinning(self):
 890         """ SNAT hairpinning """
 891
 892         host = self.pg0.remote_hosts[0]
 893         server = self.pg0.remote_hosts[1]
 894         host_in_port = 1234
 895         host_out_port = 0
 896         server_in_port = 5678
 897         server_out_port = 8765
 898
 899         self.snat_add_address(self.snat_addr)
 900         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 901         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 902                                                  is_inside=0)
 903         # add static mapping for server
 904         self.snat_add_static_mapping(server.ip4, self.snat_addr,
 905                                      server_in_port, server_out_port,
 906                                      proto=IP_PROTOS.tcp)
 907
 908         # send packet from host to server
 909         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
 910              IP(src=host.ip4, dst=self.snat_addr) /
 911              TCP(sport=host_in_port, dport=server_out_port))
 912         self.pg0.add_stream(p)
 913         self.pg_enable_capture(self.pg_interfaces)
 914         self.pg_start()
 915         capture = self.pg0.get_capture(1)
 916         p = capture[0]
 917         try:
 918             ip = p[IP]
 919             tcp = p[TCP]
 920             self.assertEqual(ip.src, self.snat_addr)
 921             self.assertEqual(ip.dst, server.ip4)
 922             self.assertNotEqual(tcp.sport, host_in_port)
 923             self.assertEqual(tcp.dport, server_in_port)
 924             host_out_port = tcp.sport
 925         except:
 926             self.logger.error(ppp("Unexpected or invalid packet:", p))
 927             raise
 928
 929         # send reply from server to host
 930         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
 931              IP(src=server.ip4, dst=self.snat_addr) /
 932              TCP(sport=server_in_port, dport=host_out_port))
 933         self.pg0.add_stream(p)
 934         self.pg_enable_capture(self.pg_interfaces)
 935         self.pg_start()
 936         capture = self.pg0.get_capture(1)
 937         p = capture[0]
 938         try:
 939             ip = p[IP]
 940             tcp = p[TCP]
 941             self.assertEqual(ip.src, self.snat_addr)
 942             self.assertEqual(ip.dst, host.ip4)
 943             self.assertEqual(tcp.sport, server_out_port)
 944             self.assertEqual(tcp.dport, host_in_port)
 945         except:
 946             self.logger.error(ppp("Unexpected or invalid packet:"), p)
 947             raise
 948
 949     def test_max_translations_per_user(self):
 950         """ MAX translations per user - recycle the least recently used """
 951
 952         self.snat_add_address(self.snat_addr)
 953         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
 954         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
 955                                                  is_inside=0)
 956
 957         # get maximum number of translations per user
 958         snat_config = self.vapi.snat_show_config()
 959
 960         # send more than maximum number of translations per user packets
 961         pkts_num = snat_config.max_translations_per_user + 5
 962         pkts = []
 963         for port in range(0, pkts_num):
 964             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
 965                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
 966                  TCP(sport=1025 + port))
 967             pkts.append(p)
 968         self.pg0.add_stream(pkts)
 969         self.pg_enable_capture(self.pg_interfaces)
 970         self.pg_start()
 971
 972         # verify number of translated packet
 973         self.pg1.get_capture(pkts_num)
 974
 975     def test_interface_addr(self):
 976         """ Acquire SNAT addresses from interface """
 977         self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
 978
 979         # no address in NAT pool
 980         adresses = self.vapi.snat_address_dump()
 981         self.assertEqual(0, len(adresses))
 982
 983         # configure interface address and check NAT address pool
 984         self.pg7.config_ip4()
 985         adresses = self.vapi.snat_address_dump()
 986         self.assertEqual(1, len(adresses))
 987         self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
 988
 989         # remove interface address and check NAT address pool
 990         self.pg7.unconfig_ip4()
 991         adresses = self.vapi.snat_address_dump()
 992         self.assertEqual(0, len(adresses))
 993
 994     def test_interface_addr_static_mapping(self):
 995         """ Static mapping with addresses from interface """
 996         self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
 997         self.snat_add_static_mapping('1.2.3.4',
 998                                      external_sw_if_index=self.pg7.sw_if_index)
 999
1000         # static mappings with external interface
1001         static_mappings = self.vapi.snat_static_mapping_dump()
1002         self.assertEqual(1, len(static_mappings))
1003         self.assertEqual(self.pg7.sw_if_index,
1004                          static_mappings[0].external_sw_if_index)
1005
1006         # configure interface address and check static mappings
1007         self.pg7.config_ip4()
1008         static_mappings = self.vapi.snat_static_mapping_dump()
1009         self.assertEqual(1, len(static_mappings))
1010         self.assertEqual(static_mappings[0].external_ip_address[0:4],
1011                          self.pg7.local_ip4n)
1012         self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index)
1013
1014         # remove interface address and check static mappings
1015         self.pg7.unconfig_ip4()
1016         static_mappings = self.vapi.snat_static_mapping_dump()
1017         self.assertEqual(0, len(static_mappings))
1018
1019     def test_ipfix_nat44_sess(self):
1020         """ S-NAT IPFIX logging NAT44 session created/delted """
1021         self.snat_add_address(self.snat_addr)
1022         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1023         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1024                                                  is_inside=0)
1025         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
1026                                      src_address=self.pg3.local_ip4n,
1027                                      path_mtu=512,
1028                                      template_interval=10)
1029         self.vapi.snat_ipfix()
1030
1031         pkts = self.create_stream_in(self.pg0, self.pg1)
1032         self.pg0.add_stream(pkts)
1033         self.pg_enable_capture(self.pg_interfaces)
1034         self.pg_start()
1035         capture = self.pg1.get_capture(len(pkts))
1036         self.verify_capture_out(capture)
1037         self.snat_add_address(self.snat_addr, is_add=0)
1038         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
1039         capture = self.pg3.get_capture(3)
1040         ipfix = IPFIXDecoder()
1041         # first load template
1042         for p in capture:
1043             self.assertTrue(p.haslayer(IPFIX))
1044             if p.haslayer(Template):
1045                 ipfix.add_template(p.getlayer(Template))
1046         # verify events in data set
1047         for p in capture:
1048             if p.haslayer(Data):
1049                 data = ipfix.decode_data_set(p.getlayer(Set))
1050                 self.verify_ipfix_nat44_ses(data)
1051
1052     def test_ipfix_addr_exhausted(self):
1053         """ S-NAT IPFIX logging NAT addresses exhausted """
1054         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1055         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1056                                                  is_inside=0)
1057         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
1058                                      src_address=self.pg3.local_ip4n,
1059                                      path_mtu=512,
1060                                      template_interval=10)
1061         self.vapi.snat_ipfix()
1062
1063         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1064              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1065              TCP(sport=3025))
1066         self.pg0.add_stream(p)
1067         self.pg_enable_capture(self.pg_interfaces)
1068         self.pg_start()
1069         capture = self.pg1.get_capture(0)
1070         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
1071         capture = self.pg3.get_capture(3)
1072         ipfix = IPFIXDecoder()
1073         # first load template
1074         for p in capture:
1075             self.assertTrue(p.haslayer(IPFIX))
1076             if p.haslayer(Template):
1077                 ipfix.add_template(p.getlayer(Template))
1078         # verify events in data set
1079         for p in capture:
1080             if p.haslayer(Data):
1081                 data = ipfix.decode_data_set(p.getlayer(Set))
1082                 self.verify_ipfix_addr_exhausted(data)
1083
1084     def test_pool_addr_fib(self):
1085         """ S-NAT add pool addresses to FIB """
1086         static_addr = '10.0.0.10'
1087         self.snat_add_address(self.snat_addr)
1088         self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
1089         self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
1090                                                  is_inside=0)
1091         self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)
1092
1093         # SNAT address
1094         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1095              ARP(op=ARP.who_has, pdst=self.snat_addr,
1096                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1097         self.pg1.add_stream(p)
1098         self.pg_enable_capture(self.pg_interfaces)
1099         self.pg_start()
1100         capture = self.pg1.get_capture(1)
1101         self.assertTrue(capture[0].haslayer(ARP))
1102         self.assertTrue(capture[0][ARP].op, ARP.is_at)
1103
1104         # 1:1 NAT address
1105         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1106              ARP(op=ARP.who_has, pdst=static_addr,
1107                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1108         self.pg1.add_stream(p)
1109         self.pg_enable_capture(self.pg_interfaces)
1110         self.pg_start()
1111         capture = self.pg1.get_capture(1)
1112         self.assertTrue(capture[0].haslayer(ARP))
1113         self.assertTrue(capture[0][ARP].op, ARP.is_at)
1114
1115         # send ARP to non-SNAT interface
1116         p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1117              ARP(op=ARP.who_has, pdst=self.snat_addr,
1118                  psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
1119         self.pg2.add_stream(p)
1120         self.pg_enable_capture(self.pg_interfaces)
1121         self.pg_start()
1122         capture = self.pg1.get_capture(0)
1123
1124         # remove addresses and verify
1125         self.snat_add_address(self.snat_addr, is_add=0)
1126         self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
1127                                      is_add=0)
1128
1129         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1130              ARP(op=ARP.who_has, pdst=self.snat_addr,
1131                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1132         self.pg1.add_stream(p)
1133         self.pg_enable_capture(self.pg_interfaces)
1134         self.pg_start()
1135         capture = self.pg1.get_capture(0)
1136
1137         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
1138              ARP(op=ARP.who_has, pdst=static_addr,
1139                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
1140         self.pg1.add_stream(p)
1141         self.pg_enable_capture(self.pg_interfaces)
1142         self.pg_start()
1143         capture = self.pg1.get_capture(0)
1144
1145     def tearDown(self):
1146         super(TestSNAT, self).tearDown()
1147         if not self.vpp_dead:
1148             self.logger.info(self.vapi.cli("show snat verbose"))
1149             self.clear_snat()
1150
1151
1152 if __name__ == '__main__':
1153     unittest.main(testRunner=VppTestRunner)