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