test/test_nat.py

   1 #!/usr/bin/env python
   2
   3 import socket
   4 import unittest
   5 import struct
   6 import StringIO
   7 import random
   8
   9 from framework import VppTestCase, VppTestRunner, running_extended_tests
  10 from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
  11 from scapy.layers.inet import IP, TCP, UDP, ICMP
  12 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
  13 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
  14 from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
  15 from scapy.layers.l2 import Ether, ARP, GRE
  16 from scapy.data import IP_PROTOS
  17 from scapy.packet import bind_layers, Raw
  18 from scapy.all import fragment6
  19 from util import ppp
  20 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
  21 from time import sleep
  22 from util import ip4_range
  23 from util import mactobinary
  24
  25
  26 class MethodHolder(VppTestCase):
  27     """ NAT create capture and verify method holder """
  28
  29     @classmethod
  30     def setUpClass(cls):
  31         super(MethodHolder, cls).setUpClass()
  32
  33     def tearDown(self):
  34         super(MethodHolder, self).tearDown()
  35
  36     def check_ip_checksum(self, pkt):
  37         """
  38         Check IP checksum of the packet
  39
  40         :param pkt: Packet to check IP checksum
  41         """
  42         new = pkt.__class__(str(pkt))
  43         del new['IP'].chksum
  44         new = new.__class__(str(new))
  45         self.assertEqual(new['IP'].chksum, pkt['IP'].chksum)
  46
  47     def check_tcp_checksum(self, pkt):
  48         """
  49         Check TCP checksum in IP packet
  50
  51         :param pkt: Packet to check TCP checksum
  52         """
  53         new = pkt.__class__(str(pkt))
  54         del new['TCP'].chksum
  55         new = new.__class__(str(new))
  56         self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum)
  57
  58     def check_udp_checksum(self, pkt):
  59         """
  60         Check UDP checksum in IP packet
  61
  62         :param pkt: Packet to check UDP checksum
  63         """
  64         new = pkt.__class__(str(pkt))
  65         del new['UDP'].chksum
  66         new = new.__class__(str(new))
  67         self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum)
  68
  69     def check_icmp_errror_embedded(self, pkt):
  70         """
  71         Check ICMP error embeded packet checksum
  72
  73         :param pkt: Packet to check ICMP error embeded packet checksum
  74         """
  75         if pkt.haslayer(IPerror):
  76             new = pkt.__class__(str(pkt))
  77             del new['IPerror'].chksum
  78             new = new.__class__(str(new))
  79             self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum)
  80
  81         if pkt.haslayer(TCPerror):
  82             new = pkt.__class__(str(pkt))
  83             del new['TCPerror'].chksum
  84             new = new.__class__(str(new))
  85             self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum)
  86
  87         if pkt.haslayer(UDPerror):
  88             if pkt['UDPerror'].chksum != 0:
  89                 new = pkt.__class__(str(pkt))
  90                 del new['UDPerror'].chksum
  91                 new = new.__class__(str(new))
  92                 self.assertEqual(new['UDPerror'].chksum,
  93                                  pkt['UDPerror'].chksum)
  94
  95         if pkt.haslayer(ICMPerror):
  96             del new['ICMPerror'].chksum
  97             new = new.__class__(str(new))
  98             self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum)
  99
 100     def check_icmp_checksum(self, pkt):
 101         """
 102         Check ICMP checksum in IPv4 packet
 103
 104         :param pkt: Packet to check ICMP checksum
 105         """
 106         new = pkt.__class__(str(pkt))
 107         del new['ICMP'].chksum
 108         new = new.__class__(str(new))
 109         self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum)
 110         if pkt.haslayer(IPerror):
 111             self.check_icmp_errror_embedded(pkt)
 112
 113     def check_icmpv6_checksum(self, pkt):
 114         """
 115         Check ICMPv6 checksum in IPv4 packet
 116
 117         :param pkt: Packet to check ICMPv6 checksum
 118         """
 119         new = pkt.__class__(str(pkt))
 120         if pkt.haslayer(ICMPv6DestUnreach):
 121             del new['ICMPv6DestUnreach'].cksum
 122             new = new.__class__(str(new))
 123             self.assertEqual(new['ICMPv6DestUnreach'].cksum,
 124                              pkt['ICMPv6DestUnreach'].cksum)
 125             self.check_icmp_errror_embedded(pkt)
 126         if pkt.haslayer(ICMPv6EchoRequest):
 127             del new['ICMPv6EchoRequest'].cksum
 128             new = new.__class__(str(new))
 129             self.assertEqual(new['ICMPv6EchoRequest'].cksum,
 130                              pkt['ICMPv6EchoRequest'].cksum)
 131         if pkt.haslayer(ICMPv6EchoReply):
 132             del new['ICMPv6EchoReply'].cksum
 133             new = new.__class__(str(new))
 134             self.assertEqual(new['ICMPv6EchoReply'].cksum,
 135                              pkt['ICMPv6EchoReply'].cksum)
 136
 137     def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
 138         """
 139         Create packet stream for inside network
 140
 141         :param in_if: Inside interface
 142         :param out_if: Outside interface
 143         :param dst_ip: Destination address
 144         :param ttl: TTL of generated packets
 145         """
 146         if dst_ip is None:
 147             dst_ip = out_if.remote_ip4
 148
 149         pkts = []
 150         # TCP
 151         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 152              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 153              TCP(sport=self.tcp_port_in, dport=20))
 154         pkts.append(p)
 155
 156         # UDP
 157         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 158              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 159              UDP(sport=self.udp_port_in, dport=20))
 160         pkts.append(p)
 161
 162         # ICMP
 163         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 164              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 165              ICMP(id=self.icmp_id_in, type='echo-request'))
 166         pkts.append(p)
 167
 168         return pkts
 169
 170     def compose_ip6(self, ip4, pref, plen):
 171         """
 172         Compose IPv4-embedded IPv6 addresses
 173
 174         :param ip4: IPv4 address
 175         :param pref: IPv6 prefix
 176         :param plen: IPv6 prefix length
 177         :returns: IPv4-embedded IPv6 addresses
 178         """
 179         pref_n = list(socket.inet_pton(socket.AF_INET6, pref))
 180         ip4_n = list(socket.inet_pton(socket.AF_INET, ip4))
 181         if plen == 32:
 182             pref_n[4] = ip4_n[0]
 183             pref_n[5] = ip4_n[1]
 184             pref_n[6] = ip4_n[2]
 185             pref_n[7] = ip4_n[3]
 186         elif plen == 40:
 187             pref_n[5] = ip4_n[0]
 188             pref_n[6] = ip4_n[1]
 189             pref_n[7] = ip4_n[2]
 190             pref_n[9] = ip4_n[3]
 191         elif plen == 48:
 192             pref_n[6] = ip4_n[0]
 193             pref_n[7] = ip4_n[1]
 194             pref_n[9] = ip4_n[2]
 195             pref_n[10] = ip4_n[3]
 196         elif plen == 56:
 197             pref_n[7] = ip4_n[0]
 198             pref_n[9] = ip4_n[1]
 199             pref_n[10] = ip4_n[2]
 200             pref_n[11] = ip4_n[3]
 201         elif plen == 64:
 202             pref_n[9] = ip4_n[0]
 203             pref_n[10] = ip4_n[1]
 204             pref_n[11] = ip4_n[2]
 205             pref_n[12] = ip4_n[3]
 206         elif plen == 96:
 207             pref_n[12] = ip4_n[0]
 208             pref_n[13] = ip4_n[1]
 209             pref_n[14] = ip4_n[2]
 210             pref_n[15] = ip4_n[3]
 211         return socket.inet_ntop(socket.AF_INET6, ''.join(pref_n))
 212
 213     def extract_ip4(self, ip6, plen):
 214         """
 215         Extract IPv4 address embedded in IPv6 addresses
 216
 217         :param ip6: IPv6 address
 218         :param plen: IPv6 prefix length
 219         :returns: extracted IPv4 address
 220         """
 221         ip6_n = list(socket.inet_pton(socket.AF_INET6, ip6))
 222         ip4_n = [None] * 4
 223         if plen == 32:
 224             ip4_n[0] = ip6_n[4]
 225             ip4_n[1] = ip6_n[5]
 226             ip4_n[2] = ip6_n[6]
 227             ip4_n[3] = ip6_n[7]
 228         elif plen == 40:
 229             ip4_n[0] = ip6_n[5]
 230             ip4_n[1] = ip6_n[6]
 231             ip4_n[2] = ip6_n[7]
 232             ip4_n[3] = ip6_n[9]
 233         elif plen == 48:
 234             ip4_n[0] = ip6_n[6]
 235             ip4_n[1] = ip6_n[7]
 236             ip4_n[2] = ip6_n[9]
 237             ip4_n[3] = ip6_n[10]
 238         elif plen == 56:
 239             ip4_n[0] = ip6_n[7]
 240             ip4_n[1] = ip6_n[9]
 241             ip4_n[2] = ip6_n[10]
 242             ip4_n[3] = ip6_n[11]
 243         elif plen == 64:
 244             ip4_n[0] = ip6_n[9]
 245             ip4_n[1] = ip6_n[10]
 246             ip4_n[2] = ip6_n[11]
 247             ip4_n[3] = ip6_n[12]
 248         elif plen == 96:
 249             ip4_n[0] = ip6_n[12]
 250             ip4_n[1] = ip6_n[13]
 251             ip4_n[2] = ip6_n[14]
 252             ip4_n[3] = ip6_n[15]
 253         return socket.inet_ntop(socket.AF_INET, ''.join(ip4_n))
 254
 255     def create_stream_in_ip6(self, in_if, out_if, hlim=64, pref=None, plen=0):
 256         """
 257         Create IPv6 packet stream for inside network
 258
 259         :param in_if: Inside interface
 260         :param out_if: Outside interface
 261         :param ttl: Hop Limit of generated packets
 262         :param pref: NAT64 prefix
 263         :param plen: NAT64 prefix length
 264         """
 265         pkts = []
 266         if pref is None:
 267             dst = ''.join(['64:ff9b::', out_if.remote_ip4])
 268         else:
 269             dst = self.compose_ip6(out_if.remote_ip4, pref, plen)
 270
 271         # TCP
 272         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 273              IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
 274              TCP(sport=self.tcp_port_in, dport=20))
 275         pkts.append(p)
 276
 277         # UDP
 278         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 279              IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
 280              UDP(sport=self.udp_port_in, dport=20))
 281         pkts.append(p)
 282
 283         # ICMP
 284         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 285              IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
 286              ICMPv6EchoRequest(id=self.icmp_id_in))
 287         pkts.append(p)
 288
 289         return pkts
 290
 291     def create_stream_out(self, out_if, dst_ip=None, ttl=64,
 292                           use_inside_ports=False):
 293         """
 294         Create packet stream for outside network
 295
 296         :param out_if: Outside interface
 297         :param dst_ip: Destination IP address (Default use global NAT address)
 298         :param ttl: TTL of generated packets
 299         :param use_inside_ports: Use inside NAT ports as destination ports
 300                instead of outside ports
 301         """
 302         if dst_ip is None:
 303             dst_ip = self.nat_addr
 304         if not use_inside_ports:
 305             tcp_port = self.tcp_port_out
 306             udp_port = self.udp_port_out
 307             icmp_id = self.icmp_id_out
 308         else:
 309             tcp_port = self.tcp_port_in
 310             udp_port = self.udp_port_in
 311             icmp_id = self.icmp_id_in
 312         pkts = []
 313         # TCP
 314         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 315              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 316              TCP(dport=tcp_port, sport=20))
 317         pkts.append(p)
 318
 319         # UDP
 320         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 321              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 322              UDP(dport=udp_port, sport=20))
 323         pkts.append(p)
 324
 325         # ICMP
 326         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 327              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 328              ICMP(id=icmp_id, type='echo-reply'))
 329         pkts.append(p)
 330
 331         return pkts
 332
 333     def create_stream_out_ip6(self, out_if, src_ip, dst_ip, hl=64):
 334         """
 335         Create packet stream for outside network
 336
 337         :param out_if: Outside interface
 338         :param dst_ip: Destination IP address (Default use global NAT address)
 339         :param hl: HL of generated packets
 340         """
 341         pkts = []
 342         # TCP
 343         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 344              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 345              TCP(dport=self.tcp_port_out, sport=20))
 346         pkts.append(p)
 347
 348         # UDP
 349         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 350              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 351              UDP(dport=self.udp_port_out, sport=20))
 352         pkts.append(p)
 353
 354         # ICMP
 355         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 356              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 357              ICMPv6EchoReply(id=self.icmp_id_out))
 358         pkts.append(p)
 359
 360         return pkts
 361
 362     def verify_capture_out(self, capture, nat_ip=None, same_port=False,
 363                            packet_num=3, dst_ip=None, is_ip6=False):
 364         """
 365         Verify captured packets on outside network
 366
 367         :param capture: Captured packets
 368         :param nat_ip: Translated IP address (Default use global NAT address)
 369         :param same_port: Sorce port number is not translated (Default False)
 370         :param packet_num: Expected number of packets (Default 3)
 371         :param dst_ip: Destination IP address (Default do not verify)
 372         :param is_ip6: If L3 protocol is IPv6 (Default False)
 373         """
 374         if is_ip6:
 375             IP46 = IPv6
 376             ICMP46 = ICMPv6EchoRequest
 377         else:
 378             IP46 = IP
 379             ICMP46 = ICMP
 380         if nat_ip is None:
 381             nat_ip = self.nat_addr
 382         self.assertEqual(packet_num, len(capture))
 383         for packet in capture:
 384             try:
 385                 if not is_ip6:
 386                     self.check_ip_checksum(packet)
 387                 self.assertEqual(packet[IP46].src, nat_ip)
 388                 if dst_ip is not None:
 389                     self.assertEqual(packet[IP46].dst, dst_ip)
 390                 if packet.haslayer(TCP):
 391                     if same_port:
 392                         self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 393                     else:
 394                         self.assertNotEqual(
 395                             packet[TCP].sport, self.tcp_port_in)
 396                     self.tcp_port_out = packet[TCP].sport
 397                     self.check_tcp_checksum(packet)
 398                 elif packet.haslayer(UDP):
 399                     if same_port:
 400                         self.assertEqual(packet[UDP].sport, self.udp_port_in)
 401                     else:
 402                         self.assertNotEqual(
 403                             packet[UDP].sport, self.udp_port_in)
 404                     self.udp_port_out = packet[UDP].sport
 405                 else:
 406                     if same_port:
 407                         self.assertEqual(packet[ICMP46].id, self.icmp_id_in)
 408                     else:
 409                         self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in)
 410                     self.icmp_id_out = packet[ICMP46].id
 411                     if is_ip6:
 412                         self.check_icmpv6_checksum(packet)
 413                     else:
 414                         self.check_icmp_checksum(packet)
 415             except:
 416                 self.logger.error(ppp("Unexpected or invalid packet "
 417                                       "(outside network):", packet))
 418                 raise
 419
 420     def verify_capture_out_ip6(self, capture, nat_ip, same_port=False,
 421                                packet_num=3, dst_ip=None):
 422         """
 423         Verify captured packets on outside network
 424
 425         :param capture: Captured packets
 426         :param nat_ip: Translated IP address
 427         :param same_port: Sorce port number is not translated (Default False)
 428         :param packet_num: Expected number of packets (Default 3)
 429         :param dst_ip: Destination IP address (Default do not verify)
 430         """
 431         return self.verify_capture_out(capture, nat_ip, same_port, packet_num,
 432                                        dst_ip, True)
 433
 434     def verify_capture_in(self, capture, in_if, packet_num=3):
 435         """
 436         Verify captured packets on inside network
 437
 438         :param capture: Captured packets
 439         :param in_if: Inside interface
 440         :param packet_num: Expected number of packets (Default 3)
 441         """
 442         self.assertEqual(packet_num, len(capture))
 443         for packet in capture:
 444             try:
 445                 self.check_ip_checksum(packet)
 446                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 447                 if packet.haslayer(TCP):
 448                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 449                     self.check_tcp_checksum(packet)
 450                 elif packet.haslayer(UDP):
 451                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 452                 else:
 453                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 454                     self.check_icmp_checksum(packet)
 455             except:
 456                 self.logger.error(ppp("Unexpected or invalid packet "
 457                                       "(inside network):", packet))
 458                 raise
 459
 460     def verify_capture_in_ip6(self, capture, src_ip, dst_ip, packet_num=3):
 461         """
 462         Verify captured IPv6 packets on inside network
 463
 464         :param capture: Captured packets
 465         :param src_ip: Source IP
 466         :param dst_ip: Destination IP address
 467         :param packet_num: Expected number of packets (Default 3)
 468         """
 469         self.assertEqual(packet_num, len(capture))
 470         for packet in capture:
 471             try:
 472                 self.assertEqual(packet[IPv6].src, src_ip)
 473                 self.assertEqual(packet[IPv6].dst, dst_ip)
 474                 if packet.haslayer(TCP):
 475                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 476                     self.check_tcp_checksum(packet)
 477                 elif packet.haslayer(UDP):
 478                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 479                     self.check_udp_checksum(packet)
 480                 else:
 481                     self.assertEqual(packet[ICMPv6EchoReply].id,
 482                                      self.icmp_id_in)
 483                     self.check_icmpv6_checksum(packet)
 484             except:
 485                 self.logger.error(ppp("Unexpected or invalid packet "
 486                                       "(inside network):", packet))
 487                 raise
 488
 489     def verify_capture_no_translation(self, capture, ingress_if, egress_if):
 490         """
 491         Verify captured packet that don't have to be translated
 492
 493         :param capture: Captured packets
 494         :param ingress_if: Ingress interface
 495         :param egress_if: Egress interface
 496         """
 497         for packet in capture:
 498             try:
 499                 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
 500                 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
 501                 if packet.haslayer(TCP):
 502                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 503                 elif packet.haslayer(UDP):
 504                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
 505                 else:
 506                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 507             except:
 508                 self.logger.error(ppp("Unexpected or invalid packet "
 509                                       "(inside network):", packet))
 510                 raise
 511
 512     def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
 513                                             packet_num=3, icmp_type=11):
 514         """
 515         Verify captured packets with ICMP errors on outside network
 516
 517         :param capture: Captured packets
 518         :param src_ip: Translated IP address or IP address of VPP
 519                        (Default use global NAT address)
 520         :param packet_num: Expected number of packets (Default 3)
 521         :param icmp_type: Type of error ICMP packet
 522                           we are expecting (Default 11)
 523         """
 524         if src_ip is None:
 525             src_ip = self.nat_addr
 526         self.assertEqual(packet_num, len(capture))
 527         for packet in capture:
 528             try:
 529                 self.assertEqual(packet[IP].src, src_ip)
 530                 self.assertTrue(packet.haslayer(ICMP))
 531                 icmp = packet[ICMP]
 532                 self.assertEqual(icmp.type, icmp_type)
 533                 self.assertTrue(icmp.haslayer(IPerror))
 534                 inner_ip = icmp[IPerror]
 535                 if inner_ip.haslayer(TCPerror):
 536                     self.assertEqual(inner_ip[TCPerror].dport,
 537                                      self.tcp_port_out)
 538                 elif inner_ip.haslayer(UDPerror):
 539                     self.assertEqual(inner_ip[UDPerror].dport,
 540                                      self.udp_port_out)
 541                 else:
 542                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
 543             except:
 544                 self.logger.error(ppp("Unexpected or invalid packet "
 545                                       "(outside network):", packet))
 546                 raise
 547
 548     def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
 549                                            icmp_type=11):
 550         """
 551         Verify captured packets with ICMP errors on inside network
 552
 553         :param capture: Captured packets
 554         :param in_if: Inside interface
 555         :param packet_num: Expected number of packets (Default 3)
 556         :param icmp_type: Type of error ICMP packet
 557                           we are expecting (Default 11)
 558         """
 559         self.assertEqual(packet_num, len(capture))
 560         for packet in capture:
 561             try:
 562                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 563                 self.assertTrue(packet.haslayer(ICMP))
 564                 icmp = packet[ICMP]
 565                 self.assertEqual(icmp.type, icmp_type)
 566                 self.assertTrue(icmp.haslayer(IPerror))
 567                 inner_ip = icmp[IPerror]
 568                 if inner_ip.haslayer(TCPerror):
 569                     self.assertEqual(inner_ip[TCPerror].sport,
 570                                      self.tcp_port_in)
 571                 elif inner_ip.haslayer(UDPerror):
 572                     self.assertEqual(inner_ip[UDPerror].sport,
 573                                      self.udp_port_in)
 574                 else:
 575                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
 576             except:
 577                 self.logger.error(ppp("Unexpected or invalid packet "
 578                                       "(inside network):", packet))
 579                 raise
 580
 581     def create_stream_frag(self, src_if, dst, sport, dport, data):
 582         """
 583         Create fragmented packet stream
 584
 585         :param src_if: Source interface
 586         :param dst: Destination IPv4 address
 587         :param sport: Source TCP port
 588         :param dport: Destination TCP port
 589         :param data: Payload data
 590         :returns: Fragmets
 591         """
 592         id = random.randint(0, 65535)
 593         p = (IP(src=src_if.remote_ip4, dst=dst) /
 594              TCP(sport=sport, dport=dport) /
 595              Raw(data))
 596         p = p.__class__(str(p))
 597         chksum = p['TCP'].chksum
 598         pkts = []
 599         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 600              IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) /
 601              TCP(sport=sport, dport=dport, chksum=chksum) /
 602              Raw(data[0:4]))
 603         pkts.append(p)
 604         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 605              IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id,
 606                 proto=IP_PROTOS.tcp) /
 607              Raw(data[4:20]))
 608         pkts.append(p)
 609         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 610              IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=IP_PROTOS.tcp,
 611                 id=id) /
 612              Raw(data[20:]))
 613         pkts.append(p)
 614         return pkts
 615
 616     def create_stream_frag_ip6(self, src_if, dst, sport, dport, data,
 617                                pref=None, plen=0, frag_size=128):
 618         """
 619         Create fragmented packet stream
 620
 621         :param src_if: Source interface
 622         :param dst: Destination IPv4 address
 623         :param sport: Source TCP port
 624         :param dport: Destination TCP port
 625         :param data: Payload data
 626         :param pref: NAT64 prefix
 627         :param plen: NAT64 prefix length
 628         :param fragsize: size of fragments
 629         :returns: Fragmets
 630         """
 631         if pref is None:
 632             dst_ip6 = ''.join(['64:ff9b::', dst])
 633         else:
 634             dst_ip6 = self.compose_ip6(dst, pref, plen)
 635
 636         p = (Ether(dst=src_if.local_mac, src=src_if.remote_mac) /
 637              IPv6(src=src_if.remote_ip6, dst=dst_ip6) /
 638              IPv6ExtHdrFragment(id=random.randint(0, 65535)) /
 639              TCP(sport=sport, dport=dport) /
 640              Raw(data))
 641
 642         return fragment6(p, frag_size)
 643
 644     def reass_frags_and_verify(self, frags, src, dst):
 645         """
 646         Reassemble and verify fragmented packet
 647
 648         :param frags: Captured fragments
 649         :param src: Source IPv4 address to verify
 650         :param dst: Destination IPv4 address to verify
 651
 652         :returns: Reassembled IPv4 packet
 653         """
 654         buffer = StringIO.StringIO()
 655         for p in frags:
 656             self.assertEqual(p[IP].src, src)
 657             self.assertEqual(p[IP].dst, dst)
 658             self.check_ip_checksum(p)
 659             buffer.seek(p[IP].frag * 8)
 660             buffer.write(p[IP].payload)
 661         ip = frags[0].getlayer(IP)
 662         ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
 663                 proto=frags[0][IP].proto)
 664         if ip.proto == IP_PROTOS.tcp:
 665             p = (ip / TCP(buffer.getvalue()))
 666             self.check_tcp_checksum(p)
 667         elif ip.proto == IP_PROTOS.udp:
 668             p = (ip / UDP(buffer.getvalue()))
 669         return p
 670
 671     def reass_frags_and_verify_ip6(self, frags, src, dst):
 672         """
 673         Reassemble and verify fragmented packet
 674
 675         :param frags: Captured fragments
 676         :param src: Source IPv6 address to verify
 677         :param dst: Destination IPv6 address to verify
 678
 679         :returns: Reassembled IPv6 packet
 680         """
 681         buffer = StringIO.StringIO()
 682         for p in frags:
 683             self.assertEqual(p[IPv6].src, src)
 684             self.assertEqual(p[IPv6].dst, dst)
 685             buffer.seek(p[IPv6ExtHdrFragment].offset * 8)
 686             buffer.write(p[IPv6ExtHdrFragment].payload)
 687         ip = IPv6(src=frags[0][IPv6].src, dst=frags[0][IPv6].dst,
 688                   nh=frags[0][IPv6ExtHdrFragment].nh)
 689         if ip.nh == IP_PROTOS.tcp:
 690             p = (ip / TCP(buffer.getvalue()))
 691             self.check_tcp_checksum(p)
 692         elif ip.nh == IP_PROTOS.udp:
 693             p = (ip / UDP(buffer.getvalue()))
 694         return p
 695
 696     def verify_ipfix_nat44_ses(self, data):
 697         """
 698         Verify IPFIX NAT44 session create/delete event
 699
 700         :param data: Decoded IPFIX data records
 701         """
 702         nat44_ses_create_num = 0
 703         nat44_ses_delete_num = 0
 704         self.assertEqual(6, len(data))
 705         for record in data:
 706             # natEvent
 707             self.assertIn(ord(record[230]), [4, 5])
 708             if ord(record[230]) == 4:
 709                 nat44_ses_create_num += 1
 710             else:
 711                 nat44_ses_delete_num += 1
 712             # sourceIPv4Address
 713             self.assertEqual(self.pg0.remote_ip4n, record[8])
 714             # postNATSourceIPv4Address
 715             self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
 716                              record[225])
 717             # ingressVRFID
 718             self.assertEqual(struct.pack("!I", 0), record[234])
 719             # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
 720             if IP_PROTOS.icmp == ord(record[4]):
 721                 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
 722                 self.assertEqual(struct.pack("!H", self.icmp_id_out),
 723                                  record[227])
 724             elif IP_PROTOS.tcp == ord(record[4]):
 725                 self.assertEqual(struct.pack("!H", self.tcp_port_in),
 726                                  record[7])
 727                 self.assertEqual(struct.pack("!H", self.tcp_port_out),
 728                                  record[227])
 729             elif IP_PROTOS.udp == ord(record[4]):
 730                 self.assertEqual(struct.pack("!H", self.udp_port_in),
 731                                  record[7])
 732                 self.assertEqual(struct.pack("!H", self.udp_port_out),
 733                                  record[227])
 734             else:
 735                 self.fail("Invalid protocol")
 736         self.assertEqual(3, nat44_ses_create_num)
 737         self.assertEqual(3, nat44_ses_delete_num)
 738
 739     def verify_ipfix_addr_exhausted(self, data):
 740         """
 741         Verify IPFIX NAT addresses event
 742
 743         :param data: Decoded IPFIX data records
 744         """
 745         self.assertEqual(1, len(data))
 746         record = data[0]
 747         # natEvent
 748         self.assertEqual(ord(record[230]), 3)
 749         # natPoolID
 750         self.assertEqual(struct.pack("!I", 0), record[283])
 751
 752     def verify_ipfix_max_sessions(self, data, limit):
 753         """
 754         Verify IPFIX maximum session entries exceeded event
 755
 756         :param data: Decoded IPFIX data records
 757         :param limit: Number of maximum session entries that can be created.
 758         """
 759         self.assertEqual(1, len(data))
 760         record = data[0]
 761         # natEvent
 762         self.assertEqual(ord(record[230]), 13)
 763         # natQuotaExceededEvent
 764         self.assertEqual(struct.pack("I", 1), record[466])
 765         # maxSessionEntries
 766         self.assertEqual(struct.pack("I", limit), record[471])
 767
 768     def verify_ipfix_max_bibs(self, data, limit):
 769         """
 770         Verify IPFIX maximum BIB entries exceeded event
 771
 772         :param data: Decoded IPFIX data records
 773         :param limit: Number of maximum BIB entries that can be created.
 774         """
 775         self.assertEqual(1, len(data))
 776         record = data[0]
 777         # natEvent
 778         self.assertEqual(ord(record[230]), 13)
 779         # natQuotaExceededEvent
 780         self.assertEqual(struct.pack("I", 2), record[466])
 781         # maxBIBEntries
 782         self.assertEqual(struct.pack("I", limit), record[472])
 783
 784     def verify_ipfix_max_fragments_ip6(self, data, limit, src_addr):
 785         """
 786         Verify IPFIX maximum IPv6 fragments pending reassembly exceeded event
 787
 788         :param data: Decoded IPFIX data records
 789         :param limit: Number of maximum fragments pending reassembly
 790         :param src_addr: IPv6 source address
 791         """
 792         self.assertEqual(1, len(data))
 793         record = data[0]
 794         # natEvent
 795         self.assertEqual(ord(record[230]), 13)
 796         # natQuotaExceededEvent
 797         self.assertEqual(struct.pack("I", 5), record[466])
 798         # maxFragmentsPendingReassembly
 799         self.assertEqual(struct.pack("I", limit), record[475])
 800         # sourceIPv6Address
 801         self.assertEqual(src_addr, record[27])
 802
 803     def verify_ipfix_max_fragments_ip4(self, data, limit, src_addr):
 804         """
 805         Verify IPFIX maximum IPv4 fragments pending reassembly exceeded event
 806
 807         :param data: Decoded IPFIX data records
 808         :param limit: Number of maximum fragments pending reassembly
 809         :param src_addr: IPv4 source address
 810         """
 811         self.assertEqual(1, len(data))
 812         record = data[0]
 813         # natEvent
 814         self.assertEqual(ord(record[230]), 13)
 815         # natQuotaExceededEvent
 816         self.assertEqual(struct.pack("I", 5), record[466])
 817         # maxFragmentsPendingReassembly
 818         self.assertEqual(struct.pack("I", limit), record[475])
 819         # sourceIPv4Address
 820         self.assertEqual(src_addr, record[8])
 821
 822     def verify_ipfix_bib(self, data, is_create, src_addr):
 823         """
 824         Verify IPFIX NAT64 BIB create and delete events
 825
 826         :param data: Decoded IPFIX data records
 827         :param is_create: Create event if nonzero value otherwise delete event
 828         :param src_addr: IPv6 source address
 829         """
 830         self.assertEqual(1, len(data))
 831         record = data[0]
 832         # natEvent
 833         if is_create:
 834             self.assertEqual(ord(record[230]), 10)
 835         else:
 836             self.assertEqual(ord(record[230]), 11)
 837         # sourceIPv6Address
 838         self.assertEqual(src_addr, record[27])
 839         # postNATSourceIPv4Address
 840         self.assertEqual(self.nat_addr_n, record[225])
 841         # protocolIdentifier
 842         self.assertEqual(IP_PROTOS.tcp, ord(record[4]))
 843         # ingressVRFID
 844         self.assertEqual(struct.pack("!I", 0), record[234])
 845         # sourceTransportPort
 846         self.assertEqual(struct.pack("!H", self.tcp_port_in), record[7])
 847         # postNAPTSourceTransportPort
 848         self.assertEqual(struct.pack("!H", self.tcp_port_out), record[227])
 849
 850     def verify_ipfix_nat64_ses(self, data, is_create, src_addr, dst_addr,
 851                                dst_port):
 852         """
 853         Verify IPFIX NAT64 session create and delete events
 854
 855         :param data: Decoded IPFIX data records
 856         :param is_create: Create event if nonzero value otherwise delete event
 857         :param src_addr: IPv6 source address
 858         :param dst_addr: IPv4 destination address
 859         :param dst_port: destination TCP port
 860         """
 861         self.assertEqual(1, len(data))
 862         record = data[0]
 863         # natEvent
 864         if is_create:
 865             self.assertEqual(ord(record[230]), 6)
 866         else:
 867             self.assertEqual(ord(record[230]), 7)
 868         # sourceIPv6Address
 869         self.assertEqual(src_addr, record[27])
 870         # destinationIPv6Address
 871         self.assertEqual(socket.inet_pton(socket.AF_INET6,
 872                                           self.compose_ip6(dst_addr,
 873                                                            '64:ff9b::',
 874                                                            96)),
 875                          record[28])
 876         # postNATSourceIPv4Address
 877         self.assertEqual(self.nat_addr_n, record[225])
 878         # postNATDestinationIPv4Address
 879         self.assertEqual(socket.inet_pton(socket.AF_INET, dst_addr),
 880                          record[226])
 881         # protocolIdentifier
 882         self.assertEqual(IP_PROTOS.tcp, ord(record[4]))
 883         # ingressVRFID
 884         self.assertEqual(struct.pack("!I", 0), record[234])
 885         # sourceTransportPort
 886         self.assertEqual(struct.pack("!H", self.tcp_port_in), record[7])
 887         # postNAPTSourceTransportPort
 888         self.assertEqual(struct.pack("!H", self.tcp_port_out), record[227])
 889         # destinationTransportPort
 890         self.assertEqual(struct.pack("!H", dst_port), record[11])
 891         # postNAPTDestinationTransportPort
 892         self.assertEqual(struct.pack("!H", dst_port), record[228])
 893
 894
 895 class TestNAT44(MethodHolder):
 896     """ NAT44 Test Cases """
 897
 898     @classmethod
 899     def setUpClass(cls):
 900         super(TestNAT44, cls).setUpClass()
 901
 902         try:
 903             cls.tcp_port_in = 6303
 904             cls.tcp_port_out = 6303
 905             cls.udp_port_in = 6304
 906             cls.udp_port_out = 6304
 907             cls.icmp_id_in = 6305
 908             cls.icmp_id_out = 6305
 909             cls.nat_addr = '10.0.0.3'
 910             cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
 911             cls.ipfix_src_port = 4739
 912             cls.ipfix_domain_id = 1
 913
 914             cls.create_pg_interfaces(range(10))
 915             cls.interfaces = list(cls.pg_interfaces[0:4])
 916
 917             for i in cls.interfaces:
 918                 i.admin_up()
 919                 i.config_ip4()
 920                 i.resolve_arp()
 921
 922             cls.pg0.generate_remote_hosts(3)
 923             cls.pg0.configure_ipv4_neighbors()
 924
 925             cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
 926             cls.vapi.ip_table_add_del(10, is_add=1)
 927             cls.vapi.ip_table_add_del(20, is_add=1)
 928
 929             cls.pg4._local_ip4 = "172.16.255.1"
 930             cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
 931             cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
 932             cls.pg4.set_table_ip4(10)
 933             cls.pg5._local_ip4 = "172.17.255.3"
 934             cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
 935             cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
 936             cls.pg5.set_table_ip4(10)
 937             cls.pg6._local_ip4 = "172.16.255.1"
 938             cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
 939             cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
 940             cls.pg6.set_table_ip4(20)
 941             for i in cls.overlapping_interfaces:
 942                 i.config_ip4()
 943                 i.admin_up()
 944                 i.resolve_arp()
 945
 946             cls.pg7.admin_up()
 947             cls.pg8.admin_up()
 948
 949             cls.pg9.generate_remote_hosts(2)
 950             cls.pg9.config_ip4()
 951             ip_addr_n = socket.inet_pton(socket.AF_INET, "10.0.0.1")
 952             cls.vapi.sw_interface_add_del_address(cls.pg9.sw_if_index,
 953                                                   ip_addr_n,
 954                                                   24)
 955             cls.pg9.admin_up()
 956             cls.pg9.resolve_arp()
 957             cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
 958             cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
 959             cls.pg9.resolve_arp()
 960
 961         except Exception:
 962             super(TestNAT44, cls).tearDownClass()
 963             raise
 964
 965     def clear_nat44(self):
 966         """
 967         Clear NAT44 configuration.
 968         """
 969         # I found no elegant way to do this
 970         self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
 971                                    dst_address_length=32,
 972                                    next_hop_address=self.pg7.remote_ip4n,
 973                                    next_hop_sw_if_index=self.pg7.sw_if_index,
 974                                    is_add=0)
 975         self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
 976                                    dst_address_length=32,
 977                                    next_hop_address=self.pg8.remote_ip4n,
 978                                    next_hop_sw_if_index=self.pg8.sw_if_index,
 979                                    is_add=0)
 980
 981         for intf in [self.pg7, self.pg8]:
 982             neighbors = self.vapi.ip_neighbor_dump(intf.sw_if_index)
 983             for n in neighbors:
 984                 self.vapi.ip_neighbor_add_del(intf.sw_if_index,
 985                                               n.mac_address,
 986                                               n.ip_address,
 987                                               is_add=0)
 988
 989         if self.pg7.has_ip4_config:
 990             self.pg7.unconfig_ip4()
 991
 992         self.vapi.nat44_forwarding_enable_disable(0)
 993
 994         interfaces = self.vapi.nat44_interface_addr_dump()
 995         for intf in interfaces:
 996             self.vapi.nat44_add_interface_addr(intf.sw_if_index,
 997                                                twice_nat=intf.twice_nat,
 998                                                is_add=0)
 999
1000         self.vapi.nat_ipfix(enable=0, src_port=self.ipfix_src_port,
1001                             domain_id=self.ipfix_domain_id)
1002         self.ipfix_src_port = 4739
1003         self.ipfix_domain_id = 1
1004
1005         interfaces = self.vapi.nat44_interface_dump()
1006         for intf in interfaces:
1007             if intf.is_inside > 1:
1008                 self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
1009                                                           0,
1010                                                           is_add=0)
1011             self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
1012                                                       intf.is_inside,
1013                                                       is_add=0)
1014
1015         interfaces = self.vapi.nat44_interface_output_feature_dump()
1016         for intf in interfaces:
1017             self.vapi.nat44_interface_add_del_output_feature(intf.sw_if_index,
1018                                                              intf.is_inside,
1019                                                              is_add=0)
1020
1021         static_mappings = self.vapi.nat44_static_mapping_dump()
1022         for sm in static_mappings:
1023             self.vapi.nat44_add_del_static_mapping(
1024                 sm.local_ip_address,
1025                 sm.external_ip_address,
1026                 local_port=sm.local_port,
1027                 external_port=sm.external_port,
1028                 addr_only=sm.addr_only,
1029                 vrf_id=sm.vrf_id,
1030                 protocol=sm.protocol,
1031                 twice_nat=sm.twice_nat,
1032                 out2in_only=sm.out2in_only,
1033                 tag=sm.tag,
1034                 is_add=0)
1035
1036         lb_static_mappings = self.vapi.nat44_lb_static_mapping_dump()
1037         for lb_sm in lb_static_mappings:
1038             self.vapi.nat44_add_del_lb_static_mapping(
1039                 lb_sm.external_addr,
1040                 lb_sm.external_port,
1041                 lb_sm.protocol,
1042                 vrf_id=lb_sm.vrf_id,
1043                 twice_nat=lb_sm.twice_nat,
1044                 out2in_only=lb_sm.out2in_only,
1045                 tag=lb_sm.tag,
1046                 is_add=0,
1047                 local_num=0,
1048                 locals=[])
1049
1050         identity_mappings = self.vapi.nat44_identity_mapping_dump()
1051         for id_m in identity_mappings:
1052             self.vapi.nat44_add_del_identity_mapping(
1053                 addr_only=id_m.addr_only,
1054                 ip=id_m.ip_address,
1055                 port=id_m.port,
1056                 sw_if_index=id_m.sw_if_index,
1057                 vrf_id=id_m.vrf_id,
1058                 protocol=id_m.protocol,
1059                 is_add=0)
1060
1061         adresses = self.vapi.nat44_address_dump()
1062         for addr in adresses:
1063             self.vapi.nat44_add_del_address_range(addr.ip_address,
1064                                                   addr.ip_address,
1065                                                   twice_nat=addr.twice_nat,
1066                                                   is_add=0)
1067
1068         self.vapi.nat_set_reass()
1069         self.vapi.nat_set_reass(is_ip6=1)
1070
1071     def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
1072                                  local_port=0, external_port=0, vrf_id=0,
1073                                  is_add=1, external_sw_if_index=0xFFFFFFFF,
1074                                  proto=0, twice_nat=0, out2in_only=0, tag=""):
1075         """
1076         Add/delete NAT44 static mapping
1077
1078         :param local_ip: Local IP address
1079         :param external_ip: External IP address
1080         :param local_port: Local port number (Optional)
1081         :param external_port: External port number (Optional)
1082         :param vrf_id: VRF ID (Default 0)
1083         :param is_add: 1 if add, 0 if delete (Default add)
1084         :param external_sw_if_index: External interface instead of IP address
1085         :param proto: IP protocol (Mandatory if port specified)
1086         :param twice_nat: 1 if translate external host address and port
1087         :param out2in_only: if 1 rule is matching only out2in direction
1088         :param tag: Opaque string tag
1089         """
1090         addr_only = 1
1091         if local_port and external_port:
1092             addr_only = 0
1093         l_ip = socket.inet_pton(socket.AF_INET, local_ip)
1094         e_ip = socket.inet_pton(socket.AF_INET, external_ip)
1095         self.vapi.nat44_add_del_static_mapping(
1096             l_ip,
1097             e_ip,
1098             external_sw_if_index,
1099             local_port,
1100             external_port,
1101             addr_only,
1102             vrf_id,
1103             proto,
1104             twice_nat,
1105             out2in_only,
1106             tag,
1107             is_add)
1108
1109     def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF, twice_nat=0):
1110         """
1111         Add/delete NAT44 address
1112
1113         :param ip: IP address
1114         :param is_add: 1 if add, 0 if delete (Default add)
1115         :param twice_nat: twice NAT address for extenal hosts
1116         """
1117         nat_addr = socket.inet_pton(socket.AF_INET, ip)
1118         self.vapi.nat44_add_del_address_range(nat_addr, nat_addr, is_add,
1119                                               vrf_id=vrf_id,
1120                                               twice_nat=twice_nat)
1121
1122     def test_dynamic(self):
1123         """ NAT44 dynamic translation test """
1124
1125         self.nat44_add_address(self.nat_addr)
1126         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1127         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1128                                                   is_inside=0)
1129
1130         # in2out
1131         pkts = self.create_stream_in(self.pg0, self.pg1)
1132         self.pg0.add_stream(pkts)
1133         self.pg_enable_capture(self.pg_interfaces)
1134         self.pg_start()
1135         capture = self.pg1.get_capture(len(pkts))
1136         self.verify_capture_out(capture)
1137
1138         # out2in
1139         pkts = self.create_stream_out(self.pg1)
1140         self.pg1.add_stream(pkts)
1141         self.pg_enable_capture(self.pg_interfaces)
1142         self.pg_start()
1143         capture = self.pg0.get_capture(len(pkts))
1144         self.verify_capture_in(capture, self.pg0)
1145
1146     def test_dynamic_icmp_errors_in2out_ttl_1(self):
1147         """ NAT44 handling of client packets with TTL=1 """
1148
1149         self.nat44_add_address(self.nat_addr)
1150         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1151         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1152                                                   is_inside=0)
1153
1154         # Client side - generate traffic
1155         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
1156         self.pg0.add_stream(pkts)
1157         self.pg_enable_capture(self.pg_interfaces)
1158         self.pg_start()
1159
1160         # Client side - verify ICMP type 11 packets
1161         capture = self.pg0.get_capture(len(pkts))
1162         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1163
1164     def test_dynamic_icmp_errors_out2in_ttl_1(self):
1165         """ NAT44 handling of server packets with TTL=1 """
1166
1167         self.nat44_add_address(self.nat_addr)
1168         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1169         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1170                                                   is_inside=0)
1171
1172         # Client side - create sessions
1173         pkts = self.create_stream_in(self.pg0, self.pg1)
1174         self.pg0.add_stream(pkts)
1175         self.pg_enable_capture(self.pg_interfaces)
1176         self.pg_start()
1177
1178         # Server side - generate traffic
1179         capture = self.pg1.get_capture(len(pkts))
1180         self.verify_capture_out(capture)
1181         pkts = self.create_stream_out(self.pg1, ttl=1)
1182         self.pg1.add_stream(pkts)
1183         self.pg_enable_capture(self.pg_interfaces)
1184         self.pg_start()
1185
1186         # Server side - verify ICMP type 11 packets
1187         capture = self.pg1.get_capture(len(pkts))
1188         self.verify_capture_out_with_icmp_errors(capture,
1189                                                  src_ip=self.pg1.local_ip4)
1190
1191     def test_dynamic_icmp_errors_in2out_ttl_2(self):
1192         """ NAT44 handling of error responses to client packets with TTL=2 """
1193
1194         self.nat44_add_address(self.nat_addr)
1195         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1196         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1197                                                   is_inside=0)
1198
1199         # Client side - generate traffic
1200         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
1201         self.pg0.add_stream(pkts)
1202         self.pg_enable_capture(self.pg_interfaces)
1203         self.pg_start()
1204
1205         # Server side - simulate ICMP type 11 response
1206         capture = self.pg1.get_capture(len(pkts))
1207         pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1208                 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1209                 ICMP(type=11) / packet[IP] for packet in capture]
1210         self.pg1.add_stream(pkts)
1211         self.pg_enable_capture(self.pg_interfaces)
1212         self.pg_start()
1213
1214         # Client side - verify ICMP type 11 packets
1215         capture = self.pg0.get_capture(len(pkts))
1216         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1217
1218     def test_dynamic_icmp_errors_out2in_ttl_2(self):
1219         """ NAT44 handling of error responses to server packets with TTL=2 """
1220
1221         self.nat44_add_address(self.nat_addr)
1222         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1223         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1224                                                   is_inside=0)
1225
1226         # Client side - create sessions
1227         pkts = self.create_stream_in(self.pg0, self.pg1)
1228         self.pg0.add_stream(pkts)
1229         self.pg_enable_capture(self.pg_interfaces)
1230         self.pg_start()
1231
1232         # Server side - generate traffic
1233         capture = self.pg1.get_capture(len(pkts))
1234         self.verify_capture_out(capture)
1235         pkts = self.create_stream_out(self.pg1, ttl=2)
1236         self.pg1.add_stream(pkts)
1237         self.pg_enable_capture(self.pg_interfaces)
1238         self.pg_start()
1239
1240         # Client side - simulate ICMP type 11 response
1241         capture = self.pg0.get_capture(len(pkts))
1242         pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1243                 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1244                 ICMP(type=11) / packet[IP] for packet in capture]
1245         self.pg0.add_stream(pkts)
1246         self.pg_enable_capture(self.pg_interfaces)
1247         self.pg_start()
1248
1249         # Server side - verify ICMP type 11 packets
1250         capture = self.pg1.get_capture(len(pkts))
1251         self.verify_capture_out_with_icmp_errors(capture)
1252
1253     def test_ping_out_interface_from_outside(self):
1254         """ Ping NAT44 out interface from outside network """
1255
1256         self.nat44_add_address(self.nat_addr)
1257         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1258         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1259                                                   is_inside=0)
1260
1261         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1262              IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
1263              ICMP(id=self.icmp_id_out, type='echo-request'))
1264         pkts = [p]
1265         self.pg1.add_stream(pkts)
1266         self.pg_enable_capture(self.pg_interfaces)
1267         self.pg_start()
1268         capture = self.pg1.get_capture(len(pkts))
1269         self.assertEqual(1, len(capture))
1270         packet = capture[0]
1271         try:
1272             self.assertEqual(packet[IP].src, self.pg1.local_ip4)
1273             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
1274             self.assertEqual(packet[ICMP].id, self.icmp_id_in)
1275             self.assertEqual(packet[ICMP].type, 0)  # echo reply
1276         except:
1277             self.logger.error(ppp("Unexpected or invalid packet "
1278                                   "(outside network):", packet))
1279             raise
1280
1281     def test_ping_internal_host_from_outside(self):
1282         """ Ping internal host from outside network """
1283
1284         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
1285         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1286         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1287                                                   is_inside=0)
1288
1289         # out2in
1290         pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1291                IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
1292                ICMP(id=self.icmp_id_out, type='echo-request'))
1293         self.pg1.add_stream(pkt)
1294         self.pg_enable_capture(self.pg_interfaces)
1295         self.pg_start()
1296         capture = self.pg0.get_capture(1)
1297         self.verify_capture_in(capture, self.pg0, packet_num=1)
1298         self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1299
1300         # in2out
1301         pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1302                IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
1303                ICMP(id=self.icmp_id_in, type='echo-reply'))
1304         self.pg0.add_stream(pkt)
1305         self.pg_enable_capture(self.pg_interfaces)
1306         self.pg_start()
1307         capture = self.pg1.get_capture(1)
1308         self.verify_capture_out(capture, same_port=True, packet_num=1)
1309         self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1310
1311     def test_forwarding(self):
1312         """ NAT44 forwarding test """
1313
1314         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1315         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1316                                                   is_inside=0)
1317         self.vapi.nat44_forwarding_enable_disable(1)
1318
1319         real_ip = self.pg0.remote_ip4n
1320         alias_ip = self.nat_addr_n
1321         self.vapi.nat44_add_del_static_mapping(local_ip=real_ip,
1322                                                external_ip=alias_ip)
1323
1324         try:
1325             # in2out - static mapping match
1326
1327             pkts = self.create_stream_out(self.pg1)
1328             self.pg1.add_stream(pkts)
1329             self.pg_enable_capture(self.pg_interfaces)
1330             self.pg_start()
1331             capture = self.pg0.get_capture(len(pkts))
1332             self.verify_capture_in(capture, self.pg0)
1333
1334             pkts = self.create_stream_in(self.pg0, self.pg1)
1335             self.pg0.add_stream(pkts)
1336             self.pg_enable_capture(self.pg_interfaces)
1337             self.pg_start()
1338             capture = self.pg1.get_capture(len(pkts))
1339             self.verify_capture_out(capture, same_port=True)
1340
1341             # in2out - no static mapping match
1342
1343             host0 = self.pg0.remote_hosts[0]
1344             self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1345             try:
1346                 pkts = self.create_stream_out(self.pg1,
1347                                               dst_ip=self.pg0.remote_ip4,
1348                                               use_inside_ports=True)
1349                 self.pg1.add_stream(pkts)
1350                 self.pg_enable_capture(self.pg_interfaces)
1351                 self.pg_start()
1352                 capture = self.pg0.get_capture(len(pkts))
1353                 self.verify_capture_in(capture, self.pg0)
1354
1355                 pkts = self.create_stream_in(self.pg0, self.pg1)
1356                 self.pg0.add_stream(pkts)
1357                 self.pg_enable_capture(self.pg_interfaces)
1358                 self.pg_start()
1359                 capture = self.pg1.get_capture(len(pkts))
1360                 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
1361                                         same_port=True)
1362             finally:
1363                 self.pg0.remote_hosts[0] = host0
1364
1365         finally:
1366             self.vapi.nat44_forwarding_enable_disable(0)
1367             self.vapi.nat44_add_del_static_mapping(local_ip=real_ip,
1368                                                    external_ip=alias_ip,
1369                                                    is_add=0)
1370
1371     def test_static_in(self):
1372         """ 1:1 NAT initialized from inside network """
1373
1374         nat_ip = "10.0.0.10"
1375         self.tcp_port_out = 6303
1376         self.udp_port_out = 6304
1377         self.icmp_id_out = 6305
1378
1379         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1380         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1381         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1382                                                   is_inside=0)
1383         sm = self.vapi.nat44_static_mapping_dump()
1384         self.assertEqual(len(sm), 1)
1385         self.assertEqual((sm[0].tag).split('\0', 1)[0], '')
1386         self.assertEqual(sm[0].protocol, 0)
1387         self.assertEqual(sm[0].local_port, 0)
1388         self.assertEqual(sm[0].external_port, 0)
1389
1390         # in2out
1391         pkts = self.create_stream_in(self.pg0, self.pg1)
1392         self.pg0.add_stream(pkts)
1393         self.pg_enable_capture(self.pg_interfaces)
1394         self.pg_start()
1395         capture = self.pg1.get_capture(len(pkts))
1396         self.verify_capture_out(capture, nat_ip, True)
1397
1398         # out2in
1399         pkts = self.create_stream_out(self.pg1, nat_ip)
1400         self.pg1.add_stream(pkts)
1401         self.pg_enable_capture(self.pg_interfaces)
1402         self.pg_start()
1403         capture = self.pg0.get_capture(len(pkts))
1404         self.verify_capture_in(capture, self.pg0)
1405
1406     def test_static_out(self):
1407         """ 1:1 NAT initialized from outside network """
1408
1409         nat_ip = "10.0.0.20"
1410         self.tcp_port_out = 6303
1411         self.udp_port_out = 6304
1412         self.icmp_id_out = 6305
1413         tag = "testTAG"
1414
1415         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
1416         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1417         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1418                                                   is_inside=0)
1419         sm = self.vapi.nat44_static_mapping_dump()
1420         self.assertEqual(len(sm), 1)
1421         self.assertEqual((sm[0].tag).split('\0', 1)[0], tag)
1422
1423         # out2in
1424         pkts = self.create_stream_out(self.pg1, nat_ip)
1425         self.pg1.add_stream(pkts)
1426         self.pg_enable_capture(self.pg_interfaces)
1427         self.pg_start()
1428         capture = self.pg0.get_capture(len(pkts))
1429         self.verify_capture_in(capture, self.pg0)
1430
1431         # in2out
1432         pkts = self.create_stream_in(self.pg0, self.pg1)
1433         self.pg0.add_stream(pkts)
1434         self.pg_enable_capture(self.pg_interfaces)
1435         self.pg_start()
1436         capture = self.pg1.get_capture(len(pkts))
1437         self.verify_capture_out(capture, nat_ip, True)
1438
1439     def test_static_with_port_in(self):
1440         """ 1:1 NAPT initialized from inside network """
1441
1442         self.tcp_port_out = 3606
1443         self.udp_port_out = 3607
1444         self.icmp_id_out = 3608
1445
1446         self.nat44_add_address(self.nat_addr)
1447         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1448                                       self.tcp_port_in, self.tcp_port_out,
1449                                       proto=IP_PROTOS.tcp)
1450         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1451                                       self.udp_port_in, self.udp_port_out,
1452                                       proto=IP_PROTOS.udp)
1453         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1454                                       self.icmp_id_in, self.icmp_id_out,
1455                                       proto=IP_PROTOS.icmp)
1456         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1457         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1458                                                   is_inside=0)
1459
1460         # in2out
1461         pkts = self.create_stream_in(self.pg0, self.pg1)
1462         self.pg0.add_stream(pkts)
1463         self.pg_enable_capture(self.pg_interfaces)
1464         self.pg_start()
1465         capture = self.pg1.get_capture(len(pkts))
1466         self.verify_capture_out(capture)
1467
1468         # out2in
1469         pkts = self.create_stream_out(self.pg1)
1470         self.pg1.add_stream(pkts)
1471         self.pg_enable_capture(self.pg_interfaces)
1472         self.pg_start()
1473         capture = self.pg0.get_capture(len(pkts))
1474         self.verify_capture_in(capture, self.pg0)
1475
1476     def test_static_with_port_out(self):
1477         """ 1:1 NAPT initialized from outside network """
1478
1479         self.tcp_port_out = 30606
1480         self.udp_port_out = 30607
1481         self.icmp_id_out = 30608
1482
1483         self.nat44_add_address(self.nat_addr)
1484         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1485                                       self.tcp_port_in, self.tcp_port_out,
1486                                       proto=IP_PROTOS.tcp)
1487         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1488                                       self.udp_port_in, self.udp_port_out,
1489                                       proto=IP_PROTOS.udp)
1490         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1491                                       self.icmp_id_in, self.icmp_id_out,
1492                                       proto=IP_PROTOS.icmp)
1493         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1494         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1495                                                   is_inside=0)
1496
1497         # out2in
1498         pkts = self.create_stream_out(self.pg1)
1499         self.pg1.add_stream(pkts)
1500         self.pg_enable_capture(self.pg_interfaces)
1501         self.pg_start()
1502         capture = self.pg0.get_capture(len(pkts))
1503         self.verify_capture_in(capture, self.pg0)
1504
1505         # in2out
1506         pkts = self.create_stream_in(self.pg0, self.pg1)
1507         self.pg0.add_stream(pkts)
1508         self.pg_enable_capture(self.pg_interfaces)
1509         self.pg_start()
1510         capture = self.pg1.get_capture(len(pkts))
1511         self.verify_capture_out(capture)
1512
1513     def test_static_with_port_out2(self):
1514         """ 1:1 NAPT symmetrical rule """
1515
1516         external_port = 80
1517         local_port = 8080
1518
1519         self.vapi.nat44_forwarding_enable_disable(1)
1520         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1521                                       local_port, external_port,
1522                                       proto=IP_PROTOS.tcp, out2in_only=1)
1523         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1524         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1525                                                   is_inside=0)
1526
1527         # from client to service
1528         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1529              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1530              TCP(sport=12345, dport=external_port))
1531         self.pg1.add_stream(p)
1532         self.pg_enable_capture(self.pg_interfaces)
1533         self.pg_start()
1534         capture = self.pg0.get_capture(1)
1535         p = capture[0]
1536         server = None
1537         try:
1538             ip = p[IP]
1539             tcp = p[TCP]
1540             self.assertEqual(ip.dst, self.pg0.remote_ip4)
1541             self.assertEqual(tcp.dport, local_port)
1542             self.check_tcp_checksum(p)
1543             self.check_ip_checksum(p)
1544         except:
1545             self.logger.error(ppp("Unexpected or invalid packet:", p))
1546             raise
1547
1548         # ICMP error
1549         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1550              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1551              ICMP(type=11) / capture[0][IP])
1552         self.pg0.add_stream(p)
1553         self.pg_enable_capture(self.pg_interfaces)
1554         self.pg_start()
1555         capture = self.pg1.get_capture(1)
1556         p = capture[0]
1557         try:
1558             self.assertEqual(p[IP].src, self.nat_addr)
1559             inner = p[IPerror]
1560             self.assertEqual(inner.dst, self.nat_addr)
1561             self.assertEqual(inner[TCPerror].dport, external_port)
1562         except:
1563             self.logger.error(ppp("Unexpected or invalid packet:", p))
1564             raise
1565
1566         # from service back to client
1567         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1568              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1569              TCP(sport=local_port, dport=12345))
1570         self.pg0.add_stream(p)
1571         self.pg_enable_capture(self.pg_interfaces)
1572         self.pg_start()
1573         capture = self.pg1.get_capture(1)
1574         p = capture[0]
1575         try:
1576             ip = p[IP]
1577             tcp = p[TCP]
1578             self.assertEqual(ip.src, self.nat_addr)
1579             self.assertEqual(tcp.sport, external_port)
1580             self.check_tcp_checksum(p)
1581             self.check_ip_checksum(p)
1582         except:
1583             self.logger.error(ppp("Unexpected or invalid packet:", p))
1584             raise
1585
1586         # ICMP error
1587         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1588              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1589              ICMP(type=11) / capture[0][IP])
1590         self.pg1.add_stream(p)
1591         self.pg_enable_capture(self.pg_interfaces)
1592         self.pg_start()
1593         capture = self.pg0.get_capture(1)
1594         p = capture[0]
1595         try:
1596             self.assertEqual(p[IP].dst, self.pg0.remote_ip4)
1597             inner = p[IPerror]
1598             self.assertEqual(inner.src, self.pg0.remote_ip4)
1599             self.assertEqual(inner[TCPerror].sport, local_port)
1600         except:
1601             self.logger.error(ppp("Unexpected or invalid packet:", p))
1602             raise
1603
1604         # from client to server (no translation)
1605         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1606              IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
1607              TCP(sport=12346, dport=local_port))
1608         self.pg1.add_stream(p)
1609         self.pg_enable_capture(self.pg_interfaces)
1610         self.pg_start()
1611         capture = self.pg0.get_capture(1)
1612         p = capture[0]
1613         server = None
1614         try:
1615             ip = p[IP]
1616             tcp = p[TCP]
1617             self.assertEqual(ip.dst, self.pg0.remote_ip4)
1618             self.assertEqual(tcp.dport, local_port)
1619             self.check_tcp_checksum(p)
1620             self.check_ip_checksum(p)
1621         except:
1622             self.logger.error(ppp("Unexpected or invalid packet:", p))
1623             raise
1624
1625         # from service back to client (no translation)
1626         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
1627              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1628              TCP(sport=local_port, dport=12346))
1629         self.pg0.add_stream(p)
1630         self.pg_enable_capture(self.pg_interfaces)
1631         self.pg_start()
1632         capture = self.pg1.get_capture(1)
1633         p = capture[0]
1634         try:
1635             ip = p[IP]
1636             tcp = p[TCP]
1637             self.assertEqual(ip.src, self.pg0.remote_ip4)
1638             self.assertEqual(tcp.sport, local_port)
1639             self.check_tcp_checksum(p)
1640             self.check_ip_checksum(p)
1641         except:
1642             self.logger.error(ppp("Unexpected or invalid packet:", p))
1643             raise
1644
1645     def test_static_vrf_aware(self):
1646         """ 1:1 NAT VRF awareness """
1647
1648         nat_ip1 = "10.0.0.30"
1649         nat_ip2 = "10.0.0.40"
1650         self.tcp_port_out = 6303
1651         self.udp_port_out = 6304
1652         self.icmp_id_out = 6305
1653
1654         self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
1655                                       vrf_id=10)
1656         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
1657                                       vrf_id=10)
1658         self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
1659                                                   is_inside=0)
1660         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1661         self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
1662
1663         # inside interface VRF match NAT44 static mapping VRF
1664         pkts = self.create_stream_in(self.pg4, self.pg3)
1665         self.pg4.add_stream(pkts)
1666         self.pg_enable_capture(self.pg_interfaces)
1667         self.pg_start()
1668         capture = self.pg3.get_capture(len(pkts))
1669         self.verify_capture_out(capture, nat_ip1, True)
1670
1671         # inside interface VRF don't match NAT44 static mapping VRF (packets
1672         # are dropped)
1673         pkts = self.create_stream_in(self.pg0, self.pg3)
1674         self.pg0.add_stream(pkts)
1675         self.pg_enable_capture(self.pg_interfaces)
1676         self.pg_start()
1677         self.pg3.assert_nothing_captured()
1678
1679     def test_dynamic_to_static(self):
1680         """ Switch from dynamic translation to 1:1NAT """
1681         nat_ip = "10.0.0.10"
1682         self.tcp_port_out = 6303
1683         self.udp_port_out = 6304
1684         self.icmp_id_out = 6305
1685
1686         self.nat44_add_address(self.nat_addr)
1687         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1688         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1689                                                   is_inside=0)
1690
1691         # dynamic
1692         pkts = self.create_stream_in(self.pg0, self.pg1)
1693         self.pg0.add_stream(pkts)
1694         self.pg_enable_capture(self.pg_interfaces)
1695         self.pg_start()
1696         capture = self.pg1.get_capture(len(pkts))
1697         self.verify_capture_out(capture)
1698
1699         # 1:1NAT
1700         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1701         sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
1702         self.assertEqual(len(sessions), 0)
1703         pkts = self.create_stream_in(self.pg0, self.pg1)
1704         self.pg0.add_stream(pkts)
1705         self.pg_enable_capture(self.pg_interfaces)
1706         self.pg_start()
1707         capture = self.pg1.get_capture(len(pkts))
1708         self.verify_capture_out(capture, nat_ip, True)
1709
1710     def test_identity_nat(self):
1711         """ Identity NAT """
1712
1713         self.vapi.nat44_add_del_identity_mapping(ip=self.pg0.remote_ip4n)
1714         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1715         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1716                                                   is_inside=0)
1717
1718         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1719              IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
1720              TCP(sport=12345, dport=56789))
1721         self.pg1.add_stream(p)
1722         self.pg_enable_capture(self.pg_interfaces)
1723         self.pg_start()
1724         capture = self.pg0.get_capture(1)
1725         p = capture[0]
1726         try:
1727             ip = p[IP]
1728             tcp = p[TCP]
1729             self.assertEqual(ip.dst, self.pg0.remote_ip4)
1730             self.assertEqual(ip.src, self.pg1.remote_ip4)
1731             self.assertEqual(tcp.dport, 56789)
1732             self.assertEqual(tcp.sport, 12345)
1733             self.check_tcp_checksum(p)
1734             self.check_ip_checksum(p)
1735         except:
1736             self.logger.error(ppp("Unexpected or invalid packet:", p))
1737             raise
1738
1739     def test_static_lb(self):
1740         """ NAT44 local service load balancing """
1741         external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
1742         external_port = 80
1743         local_port = 8080
1744         server1 = self.pg0.remote_hosts[0]
1745         server2 = self.pg0.remote_hosts[1]
1746
1747         locals = [{'addr': server1.ip4n,
1748                    'port': local_port,
1749                    'probability': 70},
1750                   {'addr': server2.ip4n,
1751                    'port': local_port,
1752                    'probability': 30}]
1753
1754         self.nat44_add_address(self.nat_addr)
1755         self.vapi.nat44_add_del_lb_static_mapping(external_addr_n,
1756                                                   external_port,
1757                                                   IP_PROTOS.tcp,
1758                                                   local_num=len(locals),
1759                                                   locals=locals)
1760         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1761         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1762                                                   is_inside=0)
1763
1764         # from client to service
1765         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1766              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1767              TCP(sport=12345, dport=external_port))
1768         self.pg1.add_stream(p)
1769         self.pg_enable_capture(self.pg_interfaces)
1770         self.pg_start()
1771         capture = self.pg0.get_capture(1)
1772         p = capture[0]
1773         server = None
1774         try:
1775             ip = p[IP]
1776             tcp = p[TCP]
1777             self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1778             if ip.dst == server1.ip4:
1779                 server = server1
1780             else:
1781                 server = server2
1782             self.assertEqual(tcp.dport, local_port)
1783             self.check_tcp_checksum(p)
1784             self.check_ip_checksum(p)
1785         except:
1786             self.logger.error(ppp("Unexpected or invalid packet:", p))
1787             raise
1788
1789         # from service back to client
1790         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1791              IP(src=server.ip4, dst=self.pg1.remote_ip4) /
1792              TCP(sport=local_port, dport=12345))
1793         self.pg0.add_stream(p)
1794         self.pg_enable_capture(self.pg_interfaces)
1795         self.pg_start()
1796         capture = self.pg1.get_capture(1)
1797         p = capture[0]
1798         try:
1799             ip = p[IP]
1800             tcp = p[TCP]
1801             self.assertEqual(ip.src, self.nat_addr)
1802             self.assertEqual(tcp.sport, external_port)
1803             self.check_tcp_checksum(p)
1804             self.check_ip_checksum(p)
1805         except:
1806             self.logger.error(ppp("Unexpected or invalid packet:", p))
1807             raise
1808
1809         # multiple clients
1810         server1_n = 0
1811         server2_n = 0
1812         clients = ip4_range(self.pg1.remote_ip4, 10, 20)
1813         pkts = []
1814         for client in clients:
1815             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1816                  IP(src=client, dst=self.nat_addr) /
1817                  TCP(sport=12345, dport=external_port))
1818             pkts.append(p)
1819         self.pg1.add_stream(pkts)
1820         self.pg_enable_capture(self.pg_interfaces)
1821         self.pg_start()
1822         capture = self.pg0.get_capture(len(pkts))
1823         for p in capture:
1824             if p[IP].dst == server1.ip4:
1825                 server1_n += 1
1826             else:
1827                 server2_n += 1
1828         self.assertTrue(server1_n > server2_n)
1829
1830     def test_static_lb_2(self):
1831         """ NAT44 local service load balancing (asymmetrical rule) """
1832         external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
1833         external_port = 80
1834         local_port = 8080
1835         server1 = self.pg0.remote_hosts[0]
1836         server2 = self.pg0.remote_hosts[1]
1837
1838         locals = [{'addr': server1.ip4n,
1839                    'port': local_port,
1840                    'probability': 70},
1841                   {'addr': server2.ip4n,
1842                    'port': local_port,
1843                    'probability': 30}]
1844
1845         self.vapi.nat44_forwarding_enable_disable(1)
1846         self.vapi.nat44_add_del_lb_static_mapping(external_addr_n,
1847                                                   external_port,
1848                                                   IP_PROTOS.tcp,
1849                                                   out2in_only=1,
1850                                                   local_num=len(locals),
1851                                                   locals=locals)
1852         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1853         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
1854                                                   is_inside=0)
1855
1856         # from client to service
1857         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1858              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1859              TCP(sport=12345, dport=external_port))
1860         self.pg1.add_stream(p)
1861         self.pg_enable_capture(self.pg_interfaces)
1862         self.pg_start()
1863         capture = self.pg0.get_capture(1)
1864         p = capture[0]
1865         server = None
1866         try:
1867             ip = p[IP]
1868             tcp = p[TCP]
1869             self.assertIn(ip.dst, [server1.ip4, server2.ip4])
1870             if ip.dst == server1.ip4:
1871                 server = server1
1872             else:
1873                 server = server2
1874             self.assertEqual(tcp.dport, local_port)
1875             self.check_tcp_checksum(p)
1876             self.check_ip_checksum(p)
1877         except:
1878             self.logger.error(ppp("Unexpected or invalid packet:", p))
1879             raise
1880
1881         # from service back to client
1882         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1883              IP(src=server.ip4, dst=self.pg1.remote_ip4) /
1884              TCP(sport=local_port, dport=12345))
1885         self.pg0.add_stream(p)
1886         self.pg_enable_capture(self.pg_interfaces)
1887         self.pg_start()
1888         capture = self.pg1.get_capture(1)
1889         p = capture[0]
1890         try:
1891             ip = p[IP]
1892             tcp = p[TCP]
1893             self.assertEqual(ip.src, self.nat_addr)
1894             self.assertEqual(tcp.sport, external_port)
1895             self.check_tcp_checksum(p)
1896             self.check_ip_checksum(p)
1897         except:
1898             self.logger.error(ppp("Unexpected or invalid packet:", p))
1899             raise
1900
1901         # from client to server (no translation)
1902         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1903              IP(src=self.pg1.remote_ip4, dst=server1.ip4) /
1904              TCP(sport=12346, dport=local_port))
1905         self.pg1.add_stream(p)
1906         self.pg_enable_capture(self.pg_interfaces)
1907         self.pg_start()
1908         capture = self.pg0.get_capture(1)
1909         p = capture[0]
1910         server = None
1911         try:
1912             ip = p[IP]
1913             tcp = p[TCP]
1914             self.assertEqual(ip.dst, server1.ip4)
1915             self.assertEqual(tcp.dport, local_port)
1916             self.check_tcp_checksum(p)
1917             self.check_ip_checksum(p)
1918         except:
1919             self.logger.error(ppp("Unexpected or invalid packet:", p))
1920             raise
1921
1922         # from service back to client (no translation)
1923         p = (Ether(src=server1.mac, dst=self.pg0.local_mac) /
1924              IP(src=server1.ip4, dst=self.pg1.remote_ip4) /
1925              TCP(sport=local_port, dport=12346))
1926         self.pg0.add_stream(p)
1927         self.pg_enable_capture(self.pg_interfaces)
1928         self.pg_start()
1929         capture = self.pg1.get_capture(1)
1930         p = capture[0]
1931         try:
1932             ip = p[IP]
1933             tcp = p[TCP]
1934             self.assertEqual(ip.src, server1.ip4)
1935             self.assertEqual(tcp.sport, local_port)
1936             self.check_tcp_checksum(p)
1937             self.check_ip_checksum(p)
1938         except:
1939             self.logger.error(ppp("Unexpected or invalid packet:", p))
1940             raise
1941
1942     def test_multiple_inside_interfaces(self):
1943         """ NAT44 multiple non-overlapping address space inside interfaces """
1944
1945         self.nat44_add_address(self.nat_addr)
1946         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
1947         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
1948         self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
1949                                                   is_inside=0)
1950
1951         # between two NAT44 inside interfaces (no translation)
1952         pkts = self.create_stream_in(self.pg0, self.pg1)
1953         self.pg0.add_stream(pkts)
1954         self.pg_enable_capture(self.pg_interfaces)
1955         self.pg_start()
1956         capture = self.pg1.get_capture(len(pkts))
1957         self.verify_capture_no_translation(capture, self.pg0, self.pg1)
1958
1959         # from NAT44 inside to interface without NAT44 feature (no translation)
1960         pkts = self.create_stream_in(self.pg0, self.pg2)
1961         self.pg0.add_stream(pkts)
1962         self.pg_enable_capture(self.pg_interfaces)
1963         self.pg_start()
1964         capture = self.pg2.get_capture(len(pkts))
1965         self.verify_capture_no_translation(capture, self.pg0, self.pg2)
1966
1967         # in2out 1st interface
1968         pkts = self.create_stream_in(self.pg0, self.pg3)
1969         self.pg0.add_stream(pkts)
1970         self.pg_enable_capture(self.pg_interfaces)
1971         self.pg_start()
1972         capture = self.pg3.get_capture(len(pkts))
1973         self.verify_capture_out(capture)
1974
1975         # out2in 1st interface
1976         pkts = self.create_stream_out(self.pg3)
1977         self.pg3.add_stream(pkts)
1978         self.pg_enable_capture(self.pg_interfaces)
1979         self.pg_start()
1980         capture = self.pg0.get_capture(len(pkts))
1981         self.verify_capture_in(capture, self.pg0)
1982
1983         # in2out 2nd interface
1984         pkts = self.create_stream_in(self.pg1, self.pg3)
1985         self.pg1.add_stream(pkts)
1986         self.pg_enable_capture(self.pg_interfaces)
1987         self.pg_start()
1988         capture = self.pg3.get_capture(len(pkts))
1989         self.verify_capture_out(capture)
1990
1991         # out2in 2nd interface
1992         pkts = self.create_stream_out(self.pg3)
1993         self.pg3.add_stream(pkts)
1994         self.pg_enable_capture(self.pg_interfaces)
1995         self.pg_start()
1996         capture = self.pg1.get_capture(len(pkts))
1997         self.verify_capture_in(capture, self.pg1)
1998
1999     def test_inside_overlapping_interfaces(self):
2000         """ NAT44 multiple inside interfaces with overlapping address space """
2001
2002         static_nat_ip = "10.0.0.10"
2003         self.nat44_add_address(self.nat_addr)
2004         self.vapi.nat44_interface_add_del_feature(self.pg3.sw_if_index,
2005                                                   is_inside=0)
2006         self.vapi.nat44_interface_add_del_feature(self.pg4.sw_if_index)
2007         self.vapi.nat44_interface_add_del_feature(self.pg5.sw_if_index)
2008         self.vapi.nat44_interface_add_del_feature(self.pg6.sw_if_index)
2009         self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
2010                                       vrf_id=20)
2011
2012         # between NAT44 inside interfaces with same VRF (no translation)
2013         pkts = self.create_stream_in(self.pg4, self.pg5)
2014         self.pg4.add_stream(pkts)
2015         self.pg_enable_capture(self.pg_interfaces)
2016         self.pg_start()
2017         capture = self.pg5.get_capture(len(pkts))
2018         self.verify_capture_no_translation(capture, self.pg4, self.pg5)
2019
2020         # between NAT44 inside interfaces with different VRF (hairpinning)
2021         p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
2022              IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
2023              TCP(sport=1234, dport=5678))
2024         self.pg4.add_stream(p)
2025         self.pg_enable_capture(self.pg_interfaces)
2026         self.pg_start()
2027         capture = self.pg6.get_capture(1)
2028         p = capture[0]
2029         try:
2030             ip = p[IP]
2031             tcp = p[TCP]
2032             self.assertEqual(ip.src, self.nat_addr)
2033             self.assertEqual(ip.dst, self.pg6.remote_ip4)
2034             self.assertNotEqual(tcp.sport, 1234)
2035             self.assertEqual(tcp.dport, 5678)
2036         except:
2037             self.logger.error(ppp("Unexpected or invalid packet:", p))
2038             raise
2039
2040         # in2out 1st interface
2041         pkts = self.create_stream_in(self.pg4, self.pg3)
2042         self.pg4.add_stream(pkts)
2043         self.pg_enable_capture(self.pg_interfaces)
2044         self.pg_start()
2045         capture = self.pg3.get_capture(len(pkts))
2046         self.verify_capture_out(capture)
2047
2048         # out2in 1st interface
2049         pkts = self.create_stream_out(self.pg3)
2050         self.pg3.add_stream(pkts)
2051         self.pg_enable_capture(self.pg_interfaces)
2052         self.pg_start()
2053         capture = self.pg4.get_capture(len(pkts))
2054         self.verify_capture_in(capture, self.pg4)
2055
2056         # in2out 2nd interface
2057         pkts = self.create_stream_in(self.pg5, self.pg3)
2058         self.pg5.add_stream(pkts)
2059         self.pg_enable_capture(self.pg_interfaces)
2060         self.pg_start()
2061         capture = self.pg3.get_capture(len(pkts))
2062         self.verify_capture_out(capture)
2063
2064         # out2in 2nd interface
2065         pkts = self.create_stream_out(self.pg3)
2066         self.pg3.add_stream(pkts)
2067         self.pg_enable_capture(self.pg_interfaces)
2068         self.pg_start()
2069         capture = self.pg5.get_capture(len(pkts))
2070         self.verify_capture_in(capture, self.pg5)
2071
2072         # pg5 session dump
2073         addresses = self.vapi.nat44_address_dump()
2074         self.assertEqual(len(addresses), 1)
2075         sessions = self.vapi.nat44_user_session_dump(self.pg5.remote_ip4n, 10)
2076         self.assertEqual(len(sessions), 3)
2077         for session in sessions:
2078             self.assertFalse(session.is_static)
2079             self.assertEqual(session.inside_ip_address[0:4],
2080                              self.pg5.remote_ip4n)
2081             self.assertEqual(session.outside_ip_address,
2082                              addresses[0].ip_address)
2083         self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
2084         self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
2085         self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
2086         self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
2087         self.assertEqual(sessions[1].inside_port, self.udp_port_in)
2088         self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
2089         self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
2090         self.assertEqual(sessions[1].outside_port, self.udp_port_out)
2091         self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
2092
2093         # in2out 3rd interface
2094         pkts = self.create_stream_in(self.pg6, self.pg3)
2095         self.pg6.add_stream(pkts)
2096         self.pg_enable_capture(self.pg_interfaces)
2097         self.pg_start()
2098         capture = self.pg3.get_capture(len(pkts))
2099         self.verify_capture_out(capture, static_nat_ip, True)
2100
2101         # out2in 3rd interface
2102         pkts = self.create_stream_out(self.pg3, static_nat_ip)
2103         self.pg3.add_stream(pkts)
2104         self.pg_enable_capture(self.pg_interfaces)
2105         self.pg_start()
2106         capture = self.pg6.get_capture(len(pkts))
2107         self.verify_capture_in(capture, self.pg6)
2108
2109         # general user and session dump verifications
2110         users = self.vapi.nat44_user_dump()
2111         self.assertTrue(len(users) >= 3)
2112         addresses = self.vapi.nat44_address_dump()
2113         self.assertEqual(len(addresses), 1)
2114         for user in users:
2115             sessions = self.vapi.nat44_user_session_dump(user.ip_address,
2116                                                          user.vrf_id)
2117             for session in sessions:
2118                 self.assertEqual(user.ip_address, session.inside_ip_address)
2119                 self.assertTrue(session.total_bytes > session.total_pkts > 0)
2120                 self.assertTrue(session.protocol in
2121                                 [IP_PROTOS.tcp, IP_PROTOS.udp,
2122                                  IP_PROTOS.icmp])
2123
2124         # pg4 session dump
2125         sessions = self.vapi.nat44_user_session_dump(self.pg4.remote_ip4n, 10)
2126         self.assertTrue(len(sessions) >= 4)
2127         for session in sessions:
2128             self.assertFalse(session.is_static)
2129             self.assertEqual(session.inside_ip_address[0:4],
2130                              self.pg4.remote_ip4n)
2131             self.assertEqual(session.outside_ip_address,
2132                              addresses[0].ip_address)
2133
2134         # pg6 session dump
2135         sessions = self.vapi.nat44_user_session_dump(self.pg6.remote_ip4n, 20)
2136         self.assertTrue(len(sessions) >= 3)
2137         for session in sessions:
2138             self.assertTrue(session.is_static)
2139             self.assertEqual(session.inside_ip_address[0:4],
2140                              self.pg6.remote_ip4n)
2141             self.assertEqual(map(ord, session.outside_ip_address[0:4]),
2142                              map(int, static_nat_ip.split('.')))
2143             self.assertTrue(session.inside_port in
2144                             [self.tcp_port_in, self.udp_port_in,
2145                              self.icmp_id_in])
2146
2147     def test_hairpinning(self):
2148         """ NAT44 hairpinning - 1:1 NAPT """
2149
2150         host = self.pg0.remote_hosts[0]
2151         server = self.pg0.remote_hosts[1]
2152         host_in_port = 1234
2153         host_out_port = 0
2154         server_in_port = 5678
2155         server_out_port = 8765
2156
2157         self.nat44_add_address(self.nat_addr)
2158         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2159         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2160                                                   is_inside=0)
2161         # add static mapping for server
2162         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
2163                                       server_in_port, server_out_port,
2164                                       proto=IP_PROTOS.tcp)
2165
2166         # send packet from host to server
2167         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2168              IP(src=host.ip4, dst=self.nat_addr) /
2169              TCP(sport=host_in_port, dport=server_out_port))
2170         self.pg0.add_stream(p)
2171         self.pg_enable_capture(self.pg_interfaces)
2172         self.pg_start()
2173         capture = self.pg0.get_capture(1)
2174         p = capture[0]
2175         try:
2176             ip = p[IP]
2177             tcp = p[TCP]
2178             self.assertEqual(ip.src, self.nat_addr)
2179             self.assertEqual(ip.dst, server.ip4)
2180             self.assertNotEqual(tcp.sport, host_in_port)
2181             self.assertEqual(tcp.dport, server_in_port)
2182             self.check_tcp_checksum(p)
2183             host_out_port = tcp.sport
2184         except:
2185             self.logger.error(ppp("Unexpected or invalid packet:", p))
2186             raise
2187
2188         # send reply from server to host
2189         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
2190              IP(src=server.ip4, dst=self.nat_addr) /
2191              TCP(sport=server_in_port, dport=host_out_port))
2192         self.pg0.add_stream(p)
2193         self.pg_enable_capture(self.pg_interfaces)
2194         self.pg_start()
2195         capture = self.pg0.get_capture(1)
2196         p = capture[0]
2197         try:
2198             ip = p[IP]
2199             tcp = p[TCP]
2200             self.assertEqual(ip.src, self.nat_addr)
2201             self.assertEqual(ip.dst, host.ip4)
2202             self.assertEqual(tcp.sport, server_out_port)
2203             self.assertEqual(tcp.dport, host_in_port)
2204             self.check_tcp_checksum(p)
2205         except:
2206             self.logger.error(ppp("Unexpected or invalid packet:", p))
2207             raise
2208
2209     def test_hairpinning2(self):
2210         """ NAT44 hairpinning - 1:1 NAT"""
2211
2212         server1_nat_ip = "10.0.0.10"
2213         server2_nat_ip = "10.0.0.11"
2214         host = self.pg0.remote_hosts[0]
2215         server1 = self.pg0.remote_hosts[1]
2216         server2 = self.pg0.remote_hosts[2]
2217         server_tcp_port = 22
2218         server_udp_port = 20
2219
2220         self.nat44_add_address(self.nat_addr)
2221         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2222         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2223                                                   is_inside=0)
2224
2225         # add static mapping for servers
2226         self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
2227         self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
2228
2229         # host to server1
2230         pkts = []
2231         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2232              IP(src=host.ip4, dst=server1_nat_ip) /
2233              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
2234         pkts.append(p)
2235         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2236              IP(src=host.ip4, dst=server1_nat_ip) /
2237              UDP(sport=self.udp_port_in, dport=server_udp_port))
2238         pkts.append(p)
2239         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2240              IP(src=host.ip4, dst=server1_nat_ip) /
2241              ICMP(id=self.icmp_id_in, type='echo-request'))
2242         pkts.append(p)
2243         self.pg0.add_stream(pkts)
2244         self.pg_enable_capture(self.pg_interfaces)
2245         self.pg_start()
2246         capture = self.pg0.get_capture(len(pkts))
2247         for packet in capture:
2248             try:
2249                 self.assertEqual(packet[IP].src, self.nat_addr)
2250                 self.assertEqual(packet[IP].dst, server1.ip4)
2251                 if packet.haslayer(TCP):
2252                     self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
2253                     self.assertEqual(packet[TCP].dport, server_tcp_port)
2254                     self.tcp_port_out = packet[TCP].sport
2255                     self.check_tcp_checksum(packet)
2256                 elif packet.haslayer(UDP):
2257                     self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
2258                     self.assertEqual(packet[UDP].dport, server_udp_port)
2259                     self.udp_port_out = packet[UDP].sport
2260                 else:
2261                     self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
2262                     self.icmp_id_out = packet[ICMP].id
2263             except:
2264                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2265                 raise
2266
2267         # server1 to host
2268         pkts = []
2269         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2270              IP(src=server1.ip4, dst=self.nat_addr) /
2271              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
2272         pkts.append(p)
2273         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2274              IP(src=server1.ip4, dst=self.nat_addr) /
2275              UDP(sport=server_udp_port, dport=self.udp_port_out))
2276         pkts.append(p)
2277         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2278              IP(src=server1.ip4, dst=self.nat_addr) /
2279              ICMP(id=self.icmp_id_out, type='echo-reply'))
2280         pkts.append(p)
2281         self.pg0.add_stream(pkts)
2282         self.pg_enable_capture(self.pg_interfaces)
2283         self.pg_start()
2284         capture = self.pg0.get_capture(len(pkts))
2285         for packet in capture:
2286             try:
2287                 self.assertEqual(packet[IP].src, server1_nat_ip)
2288                 self.assertEqual(packet[IP].dst, host.ip4)
2289                 if packet.haslayer(TCP):
2290                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
2291                     self.assertEqual(packet[TCP].sport, server_tcp_port)
2292                     self.check_tcp_checksum(packet)
2293                 elif packet.haslayer(UDP):
2294                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
2295                     self.assertEqual(packet[UDP].sport, server_udp_port)
2296                 else:
2297                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2298             except:
2299                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2300                 raise
2301
2302         # server2 to server1
2303         pkts = []
2304         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2305              IP(src=server2.ip4, dst=server1_nat_ip) /
2306              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
2307         pkts.append(p)
2308         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2309              IP(src=server2.ip4, dst=server1_nat_ip) /
2310              UDP(sport=self.udp_port_in, dport=server_udp_port))
2311         pkts.append(p)
2312         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2313              IP(src=server2.ip4, dst=server1_nat_ip) /
2314              ICMP(id=self.icmp_id_in, type='echo-request'))
2315         pkts.append(p)
2316         self.pg0.add_stream(pkts)
2317         self.pg_enable_capture(self.pg_interfaces)
2318         self.pg_start()
2319         capture = self.pg0.get_capture(len(pkts))
2320         for packet in capture:
2321             try:
2322                 self.assertEqual(packet[IP].src, server2_nat_ip)
2323                 self.assertEqual(packet[IP].dst, server1.ip4)
2324                 if packet.haslayer(TCP):
2325                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
2326                     self.assertEqual(packet[TCP].dport, server_tcp_port)
2327                     self.tcp_port_out = packet[TCP].sport
2328                     self.check_tcp_checksum(packet)
2329                 elif packet.haslayer(UDP):
2330                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
2331                     self.assertEqual(packet[UDP].dport, server_udp_port)
2332                     self.udp_port_out = packet[UDP].sport
2333                 else:
2334                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2335                     self.icmp_id_out = packet[ICMP].id
2336             except:
2337                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2338                 raise
2339
2340         # server1 to server2
2341         pkts = []
2342         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2343              IP(src=server1.ip4, dst=server2_nat_ip) /
2344              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
2345         pkts.append(p)
2346         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2347              IP(src=server1.ip4, dst=server2_nat_ip) /
2348              UDP(sport=server_udp_port, dport=self.udp_port_out))
2349         pkts.append(p)
2350         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2351              IP(src=server1.ip4, dst=server2_nat_ip) /
2352              ICMP(id=self.icmp_id_out, type='echo-reply'))
2353         pkts.append(p)
2354         self.pg0.add_stream(pkts)
2355         self.pg_enable_capture(self.pg_interfaces)
2356         self.pg_start()
2357         capture = self.pg0.get_capture(len(pkts))
2358         for packet in capture:
2359             try:
2360                 self.assertEqual(packet[IP].src, server1_nat_ip)
2361                 self.assertEqual(packet[IP].dst, server2.ip4)
2362                 if packet.haslayer(TCP):
2363                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
2364                     self.assertEqual(packet[TCP].sport, server_tcp_port)
2365                     self.check_tcp_checksum(packet)
2366                 elif packet.haslayer(UDP):
2367                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
2368                     self.assertEqual(packet[UDP].sport, server_udp_port)
2369                 else:
2370                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2371             except:
2372                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2373                 raise
2374
2375     def test_max_translations_per_user(self):
2376         """ MAX translations per user - recycle the least recently used """
2377
2378         self.nat44_add_address(self.nat_addr)
2379         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2380         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2381                                                   is_inside=0)
2382
2383         # get maximum number of translations per user
2384         nat44_config = self.vapi.nat_show_config()
2385
2386         # send more than maximum number of translations per user packets
2387         pkts_num = nat44_config.max_translations_per_user + 5
2388         pkts = []
2389         for port in range(0, pkts_num):
2390             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2391                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2392                  TCP(sport=1025 + port))
2393             pkts.append(p)
2394         self.pg0.add_stream(pkts)
2395         self.pg_enable_capture(self.pg_interfaces)
2396         self.pg_start()
2397
2398         # verify number of translated packet
2399         self.pg1.get_capture(pkts_num)
2400
2401     def test_interface_addr(self):
2402         """ Acquire NAT44 addresses from interface """
2403         self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index)
2404
2405         # no address in NAT pool
2406         adresses = self.vapi.nat44_address_dump()
2407         self.assertEqual(0, len(adresses))
2408
2409         # configure interface address and check NAT address pool
2410         self.pg7.config_ip4()
2411         adresses = self.vapi.nat44_address_dump()
2412         self.assertEqual(1, len(adresses))
2413         self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
2414
2415         # remove interface address and check NAT address pool
2416         self.pg7.unconfig_ip4()
2417         adresses = self.vapi.nat44_address_dump()
2418         self.assertEqual(0, len(adresses))
2419
2420     def test_interface_addr_static_mapping(self):
2421         """ Static mapping with addresses from interface """
2422         tag = "testTAG"
2423
2424         self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index)
2425         self.nat44_add_static_mapping(
2426             '1.2.3.4',
2427             external_sw_if_index=self.pg7.sw_if_index,
2428             tag=tag)
2429
2430         # static mappings with external interface
2431         static_mappings = self.vapi.nat44_static_mapping_dump()
2432         self.assertEqual(1, len(static_mappings))
2433         self.assertEqual(self.pg7.sw_if_index,
2434                          static_mappings[0].external_sw_if_index)
2435         self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag)
2436
2437         # configure interface address and check static mappings
2438         self.pg7.config_ip4()
2439         static_mappings = self.vapi.nat44_static_mapping_dump()
2440         self.assertEqual(1, len(static_mappings))
2441         self.assertEqual(static_mappings[0].external_ip_address[0:4],
2442                          self.pg7.local_ip4n)
2443         self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index)
2444         self.assertEqual((static_mappings[0].tag).split('\0', 1)[0], tag)
2445
2446         # remove interface address and check static mappings
2447         self.pg7.unconfig_ip4()
2448         static_mappings = self.vapi.nat44_static_mapping_dump()
2449         self.assertEqual(0, len(static_mappings))
2450
2451     def test_interface_addr_identity_nat(self):
2452         """ Identity NAT with addresses from interface """
2453
2454         port = 53053
2455         self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index)
2456         self.vapi.nat44_add_del_identity_mapping(
2457             sw_if_index=self.pg7.sw_if_index,
2458             port=port,
2459             protocol=IP_PROTOS.tcp,
2460             addr_only=0)
2461
2462         # identity mappings with external interface
2463         identity_mappings = self.vapi.nat44_identity_mapping_dump()
2464         self.assertEqual(1, len(identity_mappings))
2465         self.assertEqual(self.pg7.sw_if_index,
2466                          identity_mappings[0].sw_if_index)
2467
2468         # configure interface address and check identity mappings
2469         self.pg7.config_ip4()
2470         identity_mappings = self.vapi.nat44_identity_mapping_dump()
2471         self.assertEqual(1, len(identity_mappings))
2472         self.assertEqual(identity_mappings[0].ip_address,
2473                          self.pg7.local_ip4n)
2474         self.assertEqual(0xFFFFFFFF, identity_mappings[0].sw_if_index)
2475         self.assertEqual(port, identity_mappings[0].port)
2476         self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol)
2477
2478         # remove interface address and check identity mappings
2479         self.pg7.unconfig_ip4()
2480         identity_mappings = self.vapi.nat44_identity_mapping_dump()
2481         self.assertEqual(0, len(identity_mappings))
2482
2483     def test_ipfix_nat44_sess(self):
2484         """ IPFIX logging NAT44 session created/delted """
2485         self.ipfix_domain_id = 10
2486         self.ipfix_src_port = 20202
2487         colector_port = 30303
2488         bind_layers(UDP, IPFIX, dport=30303)
2489         self.nat44_add_address(self.nat_addr)
2490         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2491         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2492                                                   is_inside=0)
2493         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
2494                                      src_address=self.pg3.local_ip4n,
2495                                      path_mtu=512,
2496                                      template_interval=10,
2497                                      collector_port=colector_port)
2498         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
2499                             src_port=self.ipfix_src_port)
2500
2501         pkts = self.create_stream_in(self.pg0, self.pg1)
2502         self.pg0.add_stream(pkts)
2503         self.pg_enable_capture(self.pg_interfaces)
2504         self.pg_start()
2505         capture = self.pg1.get_capture(len(pkts))
2506         self.verify_capture_out(capture)
2507         self.nat44_add_address(self.nat_addr, is_add=0)
2508         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
2509         capture = self.pg3.get_capture(9)
2510         ipfix = IPFIXDecoder()
2511         # first load template
2512         for p in capture:
2513             self.assertTrue(p.haslayer(IPFIX))
2514             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2515             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2516             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2517             self.assertEqual(p[UDP].dport, colector_port)
2518             self.assertEqual(p[IPFIX].observationDomainID,
2519                              self.ipfix_domain_id)
2520             if p.haslayer(Template):
2521                 ipfix.add_template(p.getlayer(Template))
2522         # verify events in data set
2523         for p in capture:
2524             if p.haslayer(Data):
2525                 data = ipfix.decode_data_set(p.getlayer(Set))
2526                 self.verify_ipfix_nat44_ses(data)
2527
2528     def test_ipfix_addr_exhausted(self):
2529         """ IPFIX logging NAT addresses exhausted """
2530         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2531         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2532                                                   is_inside=0)
2533         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
2534                                      src_address=self.pg3.local_ip4n,
2535                                      path_mtu=512,
2536                                      template_interval=10)
2537         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
2538                             src_port=self.ipfix_src_port)
2539
2540         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2541              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2542              TCP(sport=3025))
2543         self.pg0.add_stream(p)
2544         self.pg_enable_capture(self.pg_interfaces)
2545         self.pg_start()
2546         capture = self.pg1.get_capture(0)
2547         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
2548         capture = self.pg3.get_capture(9)
2549         ipfix = IPFIXDecoder()
2550         # first load template
2551         for p in capture:
2552             self.assertTrue(p.haslayer(IPFIX))
2553             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2554             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2555             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2556             self.assertEqual(p[UDP].dport, 4739)
2557             self.assertEqual(p[IPFIX].observationDomainID,
2558                              self.ipfix_domain_id)
2559             if p.haslayer(Template):
2560                 ipfix.add_template(p.getlayer(Template))
2561         # verify events in data set
2562         for p in capture:
2563             if p.haslayer(Data):
2564                 data = ipfix.decode_data_set(p.getlayer(Set))
2565                 self.verify_ipfix_addr_exhausted(data)
2566
2567     @unittest.skipUnless(running_extended_tests(), "part of extended tests")
2568     def test_ipfix_max_sessions(self):
2569         """ IPFIX logging maximum session entries exceeded """
2570         self.nat44_add_address(self.nat_addr)
2571         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2572         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2573                                                   is_inside=0)
2574
2575         nat44_config = self.vapi.nat_show_config()
2576         max_sessions = 10 * nat44_config.translation_buckets
2577
2578         pkts = []
2579         for i in range(0, max_sessions):
2580             src = "10.10.%u.%u" % ((i & 0xFF00) >> 8, i & 0xFF)
2581             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2582                  IP(src=src, dst=self.pg1.remote_ip4) /
2583                  TCP(sport=1025))
2584             pkts.append(p)
2585         self.pg0.add_stream(pkts)
2586         self.pg_enable_capture(self.pg_interfaces)
2587         self.pg_start()
2588
2589         self.pg1.get_capture(max_sessions)
2590         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
2591                                      src_address=self.pg3.local_ip4n,
2592                                      path_mtu=512,
2593                                      template_interval=10)
2594         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
2595                             src_port=self.ipfix_src_port)
2596
2597         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2598              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2599              TCP(sport=1025))
2600         self.pg0.add_stream(p)
2601         self.pg_enable_capture(self.pg_interfaces)
2602         self.pg_start()
2603         self.pg1.get_capture(0)
2604         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
2605         capture = self.pg3.get_capture(9)
2606         ipfix = IPFIXDecoder()
2607         # first load template
2608         for p in capture:
2609             self.assertTrue(p.haslayer(IPFIX))
2610             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2611             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2612             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2613             self.assertEqual(p[UDP].dport, 4739)
2614             self.assertEqual(p[IPFIX].observationDomainID,
2615                              self.ipfix_domain_id)
2616             if p.haslayer(Template):
2617                 ipfix.add_template(p.getlayer(Template))
2618         # verify events in data set
2619         for p in capture:
2620             if p.haslayer(Data):
2621                 data = ipfix.decode_data_set(p.getlayer(Set))
2622                 self.verify_ipfix_max_sessions(data, max_sessions)
2623
2624     def test_pool_addr_fib(self):
2625         """ NAT44 add pool addresses to FIB """
2626         static_addr = '10.0.0.10'
2627         self.nat44_add_address(self.nat_addr)
2628         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2629         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2630                                                   is_inside=0)
2631         self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
2632
2633         # NAT44 address
2634         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2635              ARP(op=ARP.who_has, pdst=self.nat_addr,
2636                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2637         self.pg1.add_stream(p)
2638         self.pg_enable_capture(self.pg_interfaces)
2639         self.pg_start()
2640         capture = self.pg1.get_capture(1)
2641         self.assertTrue(capture[0].haslayer(ARP))
2642         self.assertTrue(capture[0][ARP].op, ARP.is_at)
2643
2644         # 1:1 NAT address
2645         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2646              ARP(op=ARP.who_has, pdst=static_addr,
2647                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2648         self.pg1.add_stream(p)
2649         self.pg_enable_capture(self.pg_interfaces)
2650         self.pg_start()
2651         capture = self.pg1.get_capture(1)
2652         self.assertTrue(capture[0].haslayer(ARP))
2653         self.assertTrue(capture[0][ARP].op, ARP.is_at)
2654
2655         # send ARP to non-NAT44 interface
2656         p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2657              ARP(op=ARP.who_has, pdst=self.nat_addr,
2658                  psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
2659         self.pg2.add_stream(p)
2660         self.pg_enable_capture(self.pg_interfaces)
2661         self.pg_start()
2662         capture = self.pg1.get_capture(0)
2663
2664         # remove addresses and verify
2665         self.nat44_add_address(self.nat_addr, is_add=0)
2666         self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
2667                                       is_add=0)
2668
2669         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2670              ARP(op=ARP.who_has, pdst=self.nat_addr,
2671                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2672         self.pg1.add_stream(p)
2673         self.pg_enable_capture(self.pg_interfaces)
2674         self.pg_start()
2675         capture = self.pg1.get_capture(0)
2676
2677         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2678              ARP(op=ARP.who_has, pdst=static_addr,
2679                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2680         self.pg1.add_stream(p)
2681         self.pg_enable_capture(self.pg_interfaces)
2682         self.pg_start()
2683         capture = self.pg1.get_capture(0)
2684
2685     def test_vrf_mode(self):
2686         """ NAT44 tenant VRF aware address pool mode """
2687
2688         vrf_id1 = 1
2689         vrf_id2 = 2
2690         nat_ip1 = "10.0.0.10"
2691         nat_ip2 = "10.0.0.11"
2692
2693         self.pg0.unconfig_ip4()
2694         self.pg1.unconfig_ip4()
2695         self.vapi.ip_table_add_del(vrf_id1, is_add=1)
2696         self.vapi.ip_table_add_del(vrf_id2, is_add=1)
2697         self.pg0.set_table_ip4(vrf_id1)
2698         self.pg1.set_table_ip4(vrf_id2)
2699         self.pg0.config_ip4()
2700         self.pg1.config_ip4()
2701
2702         self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
2703         self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
2704         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2705         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
2706         self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index,
2707                                                   is_inside=0)
2708
2709         # first VRF
2710         pkts = self.create_stream_in(self.pg0, self.pg2)
2711         self.pg0.add_stream(pkts)
2712         self.pg_enable_capture(self.pg_interfaces)
2713         self.pg_start()
2714         capture = self.pg2.get_capture(len(pkts))
2715         self.verify_capture_out(capture, nat_ip1)
2716
2717         # second VRF
2718         pkts = self.create_stream_in(self.pg1, self.pg2)
2719         self.pg1.add_stream(pkts)
2720         self.pg_enable_capture(self.pg_interfaces)
2721         self.pg_start()
2722         capture = self.pg2.get_capture(len(pkts))
2723         self.verify_capture_out(capture, nat_ip2)
2724
2725         self.pg0.unconfig_ip4()
2726         self.pg1.unconfig_ip4()
2727         self.pg0.set_table_ip4(0)
2728         self.pg1.set_table_ip4(0)
2729         self.vapi.ip_table_add_del(vrf_id1, is_add=0)
2730         self.vapi.ip_table_add_del(vrf_id2, is_add=0)
2731
2732     def test_vrf_feature_independent(self):
2733         """ NAT44 tenant VRF independent address pool mode """
2734
2735         nat_ip1 = "10.0.0.10"
2736         nat_ip2 = "10.0.0.11"
2737
2738         self.nat44_add_address(nat_ip1)
2739         self.nat44_add_address(nat_ip2, vrf_id=99)
2740         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2741         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index)
2742         self.vapi.nat44_interface_add_del_feature(self.pg2.sw_if_index,
2743                                                   is_inside=0)
2744
2745         # first VRF
2746         pkts = self.create_stream_in(self.pg0, self.pg2)
2747         self.pg0.add_stream(pkts)
2748         self.pg_enable_capture(self.pg_interfaces)
2749         self.pg_start()
2750         capture = self.pg2.get_capture(len(pkts))
2751         self.verify_capture_out(capture, nat_ip1)
2752
2753         # second VRF
2754         pkts = self.create_stream_in(self.pg1, self.pg2)
2755         self.pg1.add_stream(pkts)
2756         self.pg_enable_capture(self.pg_interfaces)
2757         self.pg_start()
2758         capture = self.pg2.get_capture(len(pkts))
2759         self.verify_capture_out(capture, nat_ip1)
2760
2761     def test_dynamic_ipless_interfaces(self):
2762         """ NAT44 interfaces without configured IP address """
2763
2764         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
2765                                       mactobinary(self.pg7.remote_mac),
2766                                       self.pg7.remote_ip4n,
2767                                       is_static=1)
2768         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
2769                                       mactobinary(self.pg8.remote_mac),
2770                                       self.pg8.remote_ip4n,
2771                                       is_static=1)
2772
2773         self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
2774                                    dst_address_length=32,
2775                                    next_hop_address=self.pg7.remote_ip4n,
2776                                    next_hop_sw_if_index=self.pg7.sw_if_index)
2777         self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
2778                                    dst_address_length=32,
2779                                    next_hop_address=self.pg8.remote_ip4n,
2780                                    next_hop_sw_if_index=self.pg8.sw_if_index)
2781
2782         self.nat44_add_address(self.nat_addr)
2783         self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
2784         self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
2785                                                   is_inside=0)
2786
2787         # in2out
2788         pkts = self.create_stream_in(self.pg7, self.pg8)
2789         self.pg7.add_stream(pkts)
2790         self.pg_enable_capture(self.pg_interfaces)
2791         self.pg_start()
2792         capture = self.pg8.get_capture(len(pkts))
2793         self.verify_capture_out(capture)
2794
2795         # out2in
2796         pkts = self.create_stream_out(self.pg8, self.nat_addr)
2797         self.pg8.add_stream(pkts)
2798         self.pg_enable_capture(self.pg_interfaces)
2799         self.pg_start()
2800         capture = self.pg7.get_capture(len(pkts))
2801         self.verify_capture_in(capture, self.pg7)
2802
2803     def test_static_ipless_interfaces(self):
2804         """ NAT44 interfaces without configured IP address - 1:1 NAT """
2805
2806         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
2807                                       mactobinary(self.pg7.remote_mac),
2808                                       self.pg7.remote_ip4n,
2809                                       is_static=1)
2810         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
2811                                       mactobinary(self.pg8.remote_mac),
2812                                       self.pg8.remote_ip4n,
2813                                       is_static=1)
2814
2815         self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
2816                                    dst_address_length=32,
2817                                    next_hop_address=self.pg7.remote_ip4n,
2818                                    next_hop_sw_if_index=self.pg7.sw_if_index)
2819         self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
2820                                    dst_address_length=32,
2821                                    next_hop_address=self.pg8.remote_ip4n,
2822                                    next_hop_sw_if_index=self.pg8.sw_if_index)
2823
2824         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
2825         self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
2826         self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
2827                                                   is_inside=0)
2828
2829         # out2in
2830         pkts = self.create_stream_out(self.pg8)
2831         self.pg8.add_stream(pkts)
2832         self.pg_enable_capture(self.pg_interfaces)
2833         self.pg_start()
2834         capture = self.pg7.get_capture(len(pkts))
2835         self.verify_capture_in(capture, self.pg7)
2836
2837         # in2out
2838         pkts = self.create_stream_in(self.pg7, self.pg8)
2839         self.pg7.add_stream(pkts)
2840         self.pg_enable_capture(self.pg_interfaces)
2841         self.pg_start()
2842         capture = self.pg8.get_capture(len(pkts))
2843         self.verify_capture_out(capture, self.nat_addr, True)
2844
2845     def test_static_with_port_ipless_interfaces(self):
2846         """ NAT44 interfaces without configured IP address - 1:1 NAPT """
2847
2848         self.tcp_port_out = 30606
2849         self.udp_port_out = 30607
2850         self.icmp_id_out = 30608
2851
2852         self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
2853                                       mactobinary(self.pg7.remote_mac),
2854                                       self.pg7.remote_ip4n,
2855                                       is_static=1)
2856         self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
2857                                       mactobinary(self.pg8.remote_mac),
2858                                       self.pg8.remote_ip4n,
2859                                       is_static=1)
2860
2861         self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
2862                                    dst_address_length=32,
2863                                    next_hop_address=self.pg7.remote_ip4n,
2864                                    next_hop_sw_if_index=self.pg7.sw_if_index)
2865         self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
2866                                    dst_address_length=32,
2867                                    next_hop_address=self.pg8.remote_ip4n,
2868                                    next_hop_sw_if_index=self.pg8.sw_if_index)
2869
2870         self.nat44_add_address(self.nat_addr)
2871         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2872                                       self.tcp_port_in, self.tcp_port_out,
2873                                       proto=IP_PROTOS.tcp)
2874         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2875                                       self.udp_port_in, self.udp_port_out,
2876                                       proto=IP_PROTOS.udp)
2877         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2878                                       self.icmp_id_in, self.icmp_id_out,
2879                                       proto=IP_PROTOS.icmp)
2880         self.vapi.nat44_interface_add_del_feature(self.pg7.sw_if_index)
2881         self.vapi.nat44_interface_add_del_feature(self.pg8.sw_if_index,
2882                                                   is_inside=0)
2883
2884         # out2in
2885         pkts = self.create_stream_out(self.pg8)
2886         self.pg8.add_stream(pkts)
2887         self.pg_enable_capture(self.pg_interfaces)
2888         self.pg_start()
2889         capture = self.pg7.get_capture(len(pkts))
2890         self.verify_capture_in(capture, self.pg7)
2891
2892         # in2out
2893         pkts = self.create_stream_in(self.pg7, self.pg8)
2894         self.pg7.add_stream(pkts)
2895         self.pg_enable_capture(self.pg_interfaces)
2896         self.pg_start()
2897         capture = self.pg8.get_capture(len(pkts))
2898         self.verify_capture_out(capture)
2899
2900     def test_static_unknown_proto(self):
2901         """ 1:1 NAT translate packet with unknown protocol """
2902         nat_ip = "10.0.0.10"
2903         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
2904         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2905         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2906                                                   is_inside=0)
2907
2908         # in2out
2909         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2910              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2911              GRE() /
2912              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2913              TCP(sport=1234, dport=1234))
2914         self.pg0.add_stream(p)
2915         self.pg_enable_capture(self.pg_interfaces)
2916         self.pg_start()
2917         p = self.pg1.get_capture(1)
2918         packet = p[0]
2919         try:
2920             self.assertEqual(packet[IP].src, nat_ip)
2921             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
2922             self.assertTrue(packet.haslayer(GRE))
2923             self.check_ip_checksum(packet)
2924         except:
2925             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2926             raise
2927
2928         # out2in
2929         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
2930              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
2931              GRE() /
2932              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2933              TCP(sport=1234, dport=1234))
2934         self.pg1.add_stream(p)
2935         self.pg_enable_capture(self.pg_interfaces)
2936         self.pg_start()
2937         p = self.pg0.get_capture(1)
2938         packet = p[0]
2939         try:
2940             self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
2941             self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
2942             self.assertTrue(packet.haslayer(GRE))
2943             self.check_ip_checksum(packet)
2944         except:
2945             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2946             raise
2947
2948     def test_hairpinning_static_unknown_proto(self):
2949         """ 1:1 NAT translate packet with unknown protocol - hairpinning """
2950
2951         host = self.pg0.remote_hosts[0]
2952         server = self.pg0.remote_hosts[1]
2953
2954         host_nat_ip = "10.0.0.10"
2955         server_nat_ip = "10.0.0.11"
2956
2957         self.nat44_add_static_mapping(host.ip4, host_nat_ip)
2958         self.nat44_add_static_mapping(server.ip4, server_nat_ip)
2959         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
2960         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
2961                                                   is_inside=0)
2962
2963         # host to server
2964         p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
2965              IP(src=host.ip4, dst=server_nat_ip) /
2966              GRE() /
2967              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2968              TCP(sport=1234, dport=1234))
2969         self.pg0.add_stream(p)
2970         self.pg_enable_capture(self.pg_interfaces)
2971         self.pg_start()
2972         p = self.pg0.get_capture(1)
2973         packet = p[0]
2974         try:
2975             self.assertEqual(packet[IP].src, host_nat_ip)
2976             self.assertEqual(packet[IP].dst, server.ip4)
2977             self.assertTrue(packet.haslayer(GRE))
2978             self.check_ip_checksum(packet)
2979         except:
2980             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2981             raise
2982
2983         # server to host
2984         p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
2985              IP(src=server.ip4, dst=host_nat_ip) /
2986              GRE() /
2987              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2988              TCP(sport=1234, dport=1234))
2989         self.pg0.add_stream(p)
2990         self.pg_enable_capture(self.pg_interfaces)
2991         self.pg_start()
2992         p = self.pg0.get_capture(1)
2993         packet = p[0]
2994         try:
2995             self.assertEqual(packet[IP].src, server_nat_ip)
2996             self.assertEqual(packet[IP].dst, host.ip4)
2997             self.assertTrue(packet.haslayer(GRE))
2998             self.check_ip_checksum(packet)
2999         except:
3000             self.logger.error(ppp("Unexpected or invalid packet:", packet))
3001             raise
3002
3003     def test_unknown_proto(self):
3004         """ NAT44 translate packet with unknown protocol """
3005         self.nat44_add_address(self.nat_addr)
3006         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3007         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3008                                                   is_inside=0)
3009
3010         # in2out
3011         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3012              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3013              TCP(sport=self.tcp_port_in, dport=20))
3014         self.pg0.add_stream(p)
3015         self.pg_enable_capture(self.pg_interfaces)
3016         self.pg_start()
3017         p = self.pg1.get_capture(1)
3018
3019         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3020              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3021              GRE() /
3022              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
3023              TCP(sport=1234, dport=1234))
3024         self.pg0.add_stream(p)
3025         self.pg_enable_capture(self.pg_interfaces)
3026         self.pg_start()
3027         p = self.pg1.get_capture(1)
3028         packet = p[0]
3029         try:
3030             self.assertEqual(packet[IP].src, self.nat_addr)
3031             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
3032             self.assertTrue(packet.haslayer(GRE))
3033             self.check_ip_checksum(packet)
3034         except:
3035             self.logger.error(ppp("Unexpected or invalid packet:", packet))
3036             raise
3037
3038         # out2in
3039         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
3040              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3041              GRE() /
3042              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
3043              TCP(sport=1234, dport=1234))
3044         self.pg1.add_stream(p)
3045         self.pg_enable_capture(self.pg_interfaces)
3046         self.pg_start()
3047         p = self.pg0.get_capture(1)
3048         packet = p[0]
3049         try:
3050             self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
3051             self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
3052             self.assertTrue(packet.haslayer(GRE))
3053             self.check_ip_checksum(packet)
3054         except:
3055             self.logger.error(ppp("Unexpected or invalid packet:", packet))
3056             raise
3057
3058     def test_hairpinning_unknown_proto(self):
3059         """ NAT44 translate packet with unknown protocol - hairpinning """
3060         host = self.pg0.remote_hosts[0]
3061         server = self.pg0.remote_hosts[1]
3062         host_in_port = 1234
3063         host_out_port = 0
3064         server_in_port = 5678
3065         server_out_port = 8765
3066         server_nat_ip = "10.0.0.11"
3067
3068         self.nat44_add_address(self.nat_addr)
3069         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3070         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3071                                                   is_inside=0)
3072
3073         # add static mapping for server
3074         self.nat44_add_static_mapping(server.ip4, server_nat_ip)
3075
3076         # host to server
3077         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
3078              IP(src=host.ip4, dst=server_nat_ip) /
3079              TCP(sport=host_in_port, dport=server_out_port))
3080         self.pg0.add_stream(p)
3081         self.pg_enable_capture(self.pg_interfaces)
3082         self.pg_start()
3083         capture = self.pg0.get_capture(1)
3084
3085         p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
3086              IP(src=host.ip4, dst=server_nat_ip) /
3087              GRE() /
3088              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
3089              TCP(sport=1234, dport=1234))
3090         self.pg0.add_stream(p)
3091         self.pg_enable_capture(self.pg_interfaces)
3092         self.pg_start()
3093         p = self.pg0.get_capture(1)
3094         packet = p[0]
3095         try:
3096             self.assertEqual(packet[IP].src, self.nat_addr)
3097             self.assertEqual(packet[IP].dst, server.ip4)
3098             self.assertTrue(packet.haslayer(GRE))
3099             self.check_ip_checksum(packet)
3100         except:
3101             self.logger.error(ppp("Unexpected or invalid packet:", packet))
3102             raise
3103
3104         # server to host
3105         p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
3106              IP(src=server.ip4, dst=self.nat_addr) /
3107              GRE() /
3108              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
3109              TCP(sport=1234, dport=1234))
3110         self.pg0.add_stream(p)
3111         self.pg_enable_capture(self.pg_interfaces)
3112         self.pg_start()
3113         p = self.pg0.get_capture(1)
3114         packet = p[0]
3115         try:
3116             self.assertEqual(packet[IP].src, server_nat_ip)
3117             self.assertEqual(packet[IP].dst, host.ip4)
3118             self.assertTrue(packet.haslayer(GRE))
3119             self.check_ip_checksum(packet)
3120         except:
3121             self.logger.error(ppp("Unexpected or invalid packet:", packet))
3122             raise
3123
3124     def test_output_feature(self):
3125         """ NAT44 interface output feature (in2out postrouting) """
3126         self.nat44_add_address(self.nat_addr)
3127         self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index)
3128         self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index)
3129         self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index,
3130                                                          is_inside=0)
3131
3132         # in2out
3133         pkts = self.create_stream_in(self.pg0, self.pg3)
3134         self.pg0.add_stream(pkts)
3135         self.pg_enable_capture(self.pg_interfaces)
3136         self.pg_start()
3137         capture = self.pg3.get_capture(len(pkts))
3138         self.verify_capture_out(capture)
3139
3140         # out2in
3141         pkts = self.create_stream_out(self.pg3)
3142         self.pg3.add_stream(pkts)
3143         self.pg_enable_capture(self.pg_interfaces)
3144         self.pg_start()
3145         capture = self.pg0.get_capture(len(pkts))
3146         self.verify_capture_in(capture, self.pg0)
3147
3148         # from non-NAT interface to NAT inside interface
3149         pkts = self.create_stream_in(self.pg2, self.pg0)
3150         self.pg2.add_stream(pkts)
3151         self.pg_enable_capture(self.pg_interfaces)
3152         self.pg_start()
3153         capture = self.pg0.get_capture(len(pkts))
3154         self.verify_capture_no_translation(capture, self.pg2, self.pg0)
3155
3156     def test_output_feature_vrf_aware(self):
3157         """ NAT44 interface output feature VRF aware (in2out postrouting) """
3158         nat_ip_vrf10 = "10.0.0.10"
3159         nat_ip_vrf20 = "10.0.0.20"
3160
3161         self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
3162                                    dst_address_length=32,
3163                                    next_hop_address=self.pg3.remote_ip4n,
3164                                    next_hop_sw_if_index=self.pg3.sw_if_index,
3165                                    table_id=10)
3166         self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
3167                                    dst_address_length=32,
3168                                    next_hop_address=self.pg3.remote_ip4n,
3169                                    next_hop_sw_if_index=self.pg3.sw_if_index,
3170                                    table_id=20)
3171
3172         self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
3173         self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
3174         self.vapi.nat44_interface_add_del_output_feature(self.pg4.sw_if_index)
3175         self.vapi.nat44_interface_add_del_output_feature(self.pg6.sw_if_index)
3176         self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index,
3177                                                          is_inside=0)
3178
3179         # in2out VRF 10
3180         pkts = self.create_stream_in(self.pg4, self.pg3)
3181         self.pg4.add_stream(pkts)
3182         self.pg_enable_capture(self.pg_interfaces)
3183         self.pg_start()
3184         capture = self.pg3.get_capture(len(pkts))
3185         self.verify_capture_out(capture, nat_ip=nat_ip_vrf10)
3186
3187         # out2in VRF 10
3188         pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10)
3189         self.pg3.add_stream(pkts)
3190         self.pg_enable_capture(self.pg_interfaces)
3191         self.pg_start()
3192         capture = self.pg4.get_capture(len(pkts))
3193         self.verify_capture_in(capture, self.pg4)
3194
3195         # in2out VRF 20
3196         pkts = self.create_stream_in(self.pg6, self.pg3)
3197         self.pg6.add_stream(pkts)
3198         self.pg_enable_capture(self.pg_interfaces)
3199         self.pg_start()
3200         capture = self.pg3.get_capture(len(pkts))
3201         self.verify_capture_out(capture, nat_ip=nat_ip_vrf20)
3202
3203         # out2in VRF 20
3204         pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20)
3205         self.pg3.add_stream(pkts)
3206         self.pg_enable_capture(self.pg_interfaces)
3207         self.pg_start()
3208         capture = self.pg6.get_capture(len(pkts))
3209         self.verify_capture_in(capture, self.pg6)
3210
3211     def test_output_feature_hairpinning(self):
3212         """ NAT44 interface output feature hairpinning (in2out postrouting) """
3213         host = self.pg0.remote_hosts[0]
3214         server = self.pg0.remote_hosts[1]
3215         host_in_port = 1234
3216         host_out_port = 0
3217         server_in_port = 5678
3218         server_out_port = 8765
3219
3220         self.nat44_add_address(self.nat_addr)
3221         self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index)
3222         self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index,
3223                                                          is_inside=0)
3224
3225         # add static mapping for server
3226         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
3227                                       server_in_port, server_out_port,
3228                                       proto=IP_PROTOS.tcp)
3229
3230         # send packet from host to server
3231         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
3232              IP(src=host.ip4, dst=self.nat_addr) /
3233              TCP(sport=host_in_port, dport=server_out_port))
3234         self.pg0.add_stream(p)
3235         self.pg_enable_capture(self.pg_interfaces)
3236         self.pg_start()
3237         capture = self.pg0.get_capture(1)
3238         p = capture[0]
3239         try:
3240             ip = p[IP]
3241             tcp = p[TCP]
3242             self.assertEqual(ip.src, self.nat_addr)
3243             self.assertEqual(ip.dst, server.ip4)
3244             self.assertNotEqual(tcp.sport, host_in_port)
3245             self.assertEqual(tcp.dport, server_in_port)
3246             self.check_tcp_checksum(p)
3247             host_out_port = tcp.sport
3248         except:
3249             self.logger.error(ppp("Unexpected or invalid packet:", p))
3250             raise
3251
3252         # send reply from server to host
3253         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
3254              IP(src=server.ip4, dst=self.nat_addr) /
3255              TCP(sport=server_in_port, dport=host_out_port))
3256         self.pg0.add_stream(p)
3257         self.pg_enable_capture(self.pg_interfaces)
3258         self.pg_start()
3259         capture = self.pg0.get_capture(1)
3260         p = capture[0]
3261         try:
3262             ip = p[IP]
3263             tcp = p[TCP]
3264             self.assertEqual(ip.src, self.nat_addr)
3265             self.assertEqual(ip.dst, host.ip4)
3266             self.assertEqual(tcp.sport, server_out_port)
3267             self.assertEqual(tcp.dport, host_in_port)
3268             self.check_tcp_checksum(p)
3269         except:
3270             self.logger.error(ppp("Unexpected or invalid packet:", p))
3271             raise
3272
3273     def test_one_armed_nat44(self):
3274         """ One armed NAT44 """
3275         remote_host = self.pg9.remote_hosts[0]
3276         local_host = self.pg9.remote_hosts[1]
3277         external_port = 0
3278
3279         self.nat44_add_address(self.nat_addr)
3280         self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index)
3281         self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index,
3282                                                   is_inside=0)
3283
3284         # in2out
3285         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3286              IP(src=local_host.ip4, dst=remote_host.ip4) /
3287              TCP(sport=12345, dport=80))
3288         self.pg9.add_stream(p)
3289         self.pg_enable_capture(self.pg_interfaces)
3290         self.pg_start()
3291         capture = self.pg9.get_capture(1)
3292         p = capture[0]
3293         try:
3294             ip = p[IP]
3295             tcp = p[TCP]
3296             self.assertEqual(ip.src, self.nat_addr)
3297             self.assertEqual(ip.dst, remote_host.ip4)
3298             self.assertNotEqual(tcp.sport, 12345)
3299             external_port = tcp.sport
3300             self.assertEqual(tcp.dport, 80)
3301             self.check_tcp_checksum(p)
3302             self.check_ip_checksum(p)
3303         except:
3304             self.logger.error(ppp("Unexpected or invalid packet:", p))
3305             raise
3306
3307         # out2in
3308         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3309              IP(src=remote_host.ip4, dst=self.nat_addr) /
3310              TCP(sport=80, dport=external_port))
3311         self.pg9.add_stream(p)
3312         self.pg_enable_capture(self.pg_interfaces)
3313         self.pg_start()
3314         capture = self.pg9.get_capture(1)
3315         p = capture[0]
3316         try:
3317             ip = p[IP]
3318             tcp = p[TCP]
3319             self.assertEqual(ip.src, remote_host.ip4)
3320             self.assertEqual(ip.dst, local_host.ip4)
3321             self.assertEqual(tcp.sport, 80)
3322             self.assertEqual(tcp.dport, 12345)
3323             self.check_tcp_checksum(p)
3324             self.check_ip_checksum(p)
3325         except:
3326             self.logger.error(ppp("Unexpected or invalid packet:", p))
3327             raise
3328
3329     def test_one_armed_nat44_static(self):
3330         """ One armed NAT44 and 1:1 NAPT symmetrical rule """
3331         remote_host = self.pg9.remote_hosts[0]
3332         local_host = self.pg9.remote_hosts[1]
3333         external_port = 80
3334         local_port = 8080
3335         eh_port_in = 0
3336
3337         self.vapi.nat44_forwarding_enable_disable(1)
3338         self.nat44_add_address(self.nat_addr, twice_nat=1)
3339         self.nat44_add_static_mapping(local_host.ip4, self.nat_addr,
3340                                       local_port, external_port,
3341                                       proto=IP_PROTOS.tcp, out2in_only=1,
3342                                       twice_nat=1)
3343         self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index)
3344         self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index,
3345                                                   is_inside=0)
3346
3347         # from client to service
3348         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3349              IP(src=remote_host.ip4, dst=self.nat_addr) /
3350              TCP(sport=12345, dport=external_port))
3351         self.pg9.add_stream(p)
3352         self.pg_enable_capture(self.pg_interfaces)
3353         self.pg_start()
3354         capture = self.pg9.get_capture(1)
3355         p = capture[0]
3356         server = None
3357         try:
3358             ip = p[IP]
3359             tcp = p[TCP]
3360             self.assertEqual(ip.dst, local_host.ip4)
3361             self.assertEqual(ip.src, self.nat_addr)
3362             self.assertEqual(tcp.dport, local_port)
3363             self.assertNotEqual(tcp.sport, 12345)
3364             eh_port_in = tcp.sport
3365             self.check_tcp_checksum(p)
3366             self.check_ip_checksum(p)
3367         except:
3368             self.logger.error(ppp("Unexpected or invalid packet:", p))
3369             raise
3370
3371         # from service back to client
3372         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3373              IP(src=local_host.ip4, dst=self.nat_addr) /
3374              TCP(sport=local_port, dport=eh_port_in))
3375         self.pg9.add_stream(p)
3376         self.pg_enable_capture(self.pg_interfaces)
3377         self.pg_start()
3378         capture = self.pg9.get_capture(1)
3379         p = capture[0]
3380         try:
3381             ip = p[IP]
3382             tcp = p[TCP]
3383             self.assertEqual(ip.src, self.nat_addr)
3384             self.assertEqual(ip.dst, remote_host.ip4)
3385             self.assertEqual(tcp.sport, external_port)
3386             self.assertEqual(tcp.dport, 12345)
3387             self.check_tcp_checksum(p)
3388             self.check_ip_checksum(p)
3389         except:
3390             self.logger.error(ppp("Unexpected or invalid packet:", p))
3391             raise
3392
3393     def test_del_session(self):
3394         """ Delete NAT44 session """
3395         self.nat44_add_address(self.nat_addr)
3396         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3397         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3398                                                   is_inside=0)
3399
3400         pkts = self.create_stream_in(self.pg0, self.pg1)
3401         self.pg0.add_stream(pkts)
3402         self.pg_enable_capture(self.pg_interfaces)
3403         self.pg_start()
3404         capture = self.pg1.get_capture(len(pkts))
3405
3406         sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
3407         nsessions = len(sessions)
3408
3409         self.vapi.nat44_del_session(sessions[0].inside_ip_address,
3410                                     sessions[0].inside_port,
3411                                     sessions[0].protocol)
3412         self.vapi.nat44_del_session(sessions[1].outside_ip_address,
3413                                     sessions[1].outside_port,
3414                                     sessions[1].protocol,
3415                                     is_in=0)
3416
3417         sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
3418         self.assertEqual(nsessions - len(sessions), 2)
3419
3420     def test_set_get_reass(self):
3421         """ NAT44 set/get virtual fragmentation reassembly """
3422         reas_cfg1 = self.vapi.nat_get_reass()
3423
3424         self.vapi.nat_set_reass(timeout=reas_cfg1.ip4_timeout + 5,
3425                                 max_reass=reas_cfg1.ip4_max_reass * 2,
3426                                 max_frag=reas_cfg1.ip4_max_frag * 2)
3427
3428         reas_cfg2 = self.vapi.nat_get_reass()
3429
3430         self.assertEqual(reas_cfg1.ip4_timeout + 5, reas_cfg2.ip4_timeout)
3431         self.assertEqual(reas_cfg1.ip4_max_reass * 2, reas_cfg2.ip4_max_reass)
3432         self.assertEqual(reas_cfg1.ip4_max_frag * 2, reas_cfg2.ip4_max_frag)
3433
3434         self.vapi.nat_set_reass(drop_frag=1)
3435         self.assertTrue(self.vapi.nat_get_reass().ip4_drop_frag)
3436
3437     def test_frag_in_order(self):
3438         """ NAT44 translate fragments arriving in order """
3439         self.nat44_add_address(self.nat_addr)
3440         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3441         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3442                                                   is_inside=0)
3443
3444         data = "A" * 4 + "B" * 16 + "C" * 3
3445         self.tcp_port_in = random.randint(1025, 65535)
3446
3447         reass = self.vapi.nat_reass_dump()
3448         reass_n_start = len(reass)
3449
3450         # in2out
3451         pkts = self.create_stream_frag(self.pg0,
3452                                        self.pg1.remote_ip4,
3453                                        self.tcp_port_in,
3454                                        20,
3455                                        data)
3456         self.pg0.add_stream(pkts)
3457         self.pg_enable_capture(self.pg_interfaces)
3458         self.pg_start()
3459         frags = self.pg1.get_capture(len(pkts))
3460         p = self.reass_frags_and_verify(frags,
3461                                         self.nat_addr,
3462                                         self.pg1.remote_ip4)
3463         self.assertEqual(p[TCP].dport, 20)
3464         self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
3465         self.tcp_port_out = p[TCP].sport
3466         self.assertEqual(data, p[Raw].load)
3467
3468         # out2in
3469         pkts = self.create_stream_frag(self.pg1,
3470                                        self.nat_addr,
3471                                        20,
3472                                        self.tcp_port_out,
3473                                        data)
3474         self.pg1.add_stream(pkts)
3475         self.pg_enable_capture(self.pg_interfaces)
3476         self.pg_start()
3477         frags = self.pg0.get_capture(len(pkts))
3478         p = self.reass_frags_and_verify(frags,
3479                                         self.pg1.remote_ip4,
3480                                         self.pg0.remote_ip4)
3481         self.assertEqual(p[TCP].sport, 20)
3482         self.assertEqual(p[TCP].dport, self.tcp_port_in)
3483         self.assertEqual(data, p[Raw].load)
3484
3485         reass = self.vapi.nat_reass_dump()
3486         reass_n_end = len(reass)
3487
3488         self.assertEqual(reass_n_end - reass_n_start, 2)
3489
3490     def test_reass_hairpinning(self):
3491         """ NAT44 fragments hairpinning """
3492         host = self.pg0.remote_hosts[0]
3493         server = self.pg0.remote_hosts[1]
3494         host_in_port = random.randint(1025, 65535)
3495         host_out_port = 0
3496         server_in_port = random.randint(1025, 65535)
3497         server_out_port = random.randint(1025, 65535)
3498         data = "A" * 4 + "B" * 16 + "C" * 3
3499
3500         self.nat44_add_address(self.nat_addr)
3501         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3502         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3503                                                   is_inside=0)
3504         # add static mapping for server
3505         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
3506                                       server_in_port, server_out_port,
3507                                       proto=IP_PROTOS.tcp)
3508
3509         # send packet from host to server
3510         pkts = self.create_stream_frag(self.pg0,
3511                                        self.nat_addr,
3512                                        host_in_port,
3513                                        server_out_port,
3514                                        data)
3515         self.pg0.add_stream(pkts)
3516         self.pg_enable_capture(self.pg_interfaces)
3517         self.pg_start()
3518         frags = self.pg0.get_capture(len(pkts))
3519         p = self.reass_frags_and_verify(frags,
3520                                         self.nat_addr,
3521                                         server.ip4)
3522         self.assertNotEqual(p[TCP].sport, host_in_port)
3523         self.assertEqual(p[TCP].dport, server_in_port)
3524         self.assertEqual(data, p[Raw].load)
3525
3526     def test_frag_out_of_order(self):
3527         """ NAT44 translate fragments arriving out of order """
3528         self.nat44_add_address(self.nat_addr)
3529         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3530         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3531                                                   is_inside=0)
3532
3533         data = "A" * 4 + "B" * 16 + "C" * 3
3534         random.randint(1025, 65535)
3535
3536         # in2out
3537         pkts = self.create_stream_frag(self.pg0,
3538                                        self.pg1.remote_ip4,
3539                                        self.tcp_port_in,
3540                                        20,
3541                                        data)
3542         pkts.reverse()
3543         self.pg0.add_stream(pkts)
3544         self.pg_enable_capture(self.pg_interfaces)
3545         self.pg_start()
3546         frags = self.pg1.get_capture(len(pkts))
3547         p = self.reass_frags_and_verify(frags,
3548                                         self.nat_addr,
3549                                         self.pg1.remote_ip4)
3550         self.assertEqual(p[TCP].dport, 20)
3551         self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
3552         self.tcp_port_out = p[TCP].sport
3553         self.assertEqual(data, p[Raw].load)
3554
3555         # out2in
3556         pkts = self.create_stream_frag(self.pg1,
3557                                        self.nat_addr,
3558                                        20,
3559                                        self.tcp_port_out,
3560                                        data)
3561         pkts.reverse()
3562         self.pg1.add_stream(pkts)
3563         self.pg_enable_capture(self.pg_interfaces)
3564         self.pg_start()
3565         frags = self.pg0.get_capture(len(pkts))
3566         p = self.reass_frags_and_verify(frags,
3567                                         self.pg1.remote_ip4,
3568                                         self.pg0.remote_ip4)
3569         self.assertEqual(p[TCP].sport, 20)
3570         self.assertEqual(p[TCP].dport, self.tcp_port_in)
3571         self.assertEqual(data, p[Raw].load)
3572
3573     def test_port_restricted(self):
3574         """ Port restricted NAT44 (MAP-E CE) """
3575         self.nat44_add_address(self.nat_addr)
3576         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3577         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3578                                                   is_inside=0)
3579         self.vapi.cli("nat addr-port-assignment-alg map-e psid 10 "
3580                       "psid-offset 6 psid-len 6")
3581
3582         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3583              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3584              TCP(sport=4567, dport=22))
3585         self.pg0.add_stream(p)
3586         self.pg_enable_capture(self.pg_interfaces)
3587         self.pg_start()
3588         capture = self.pg1.get_capture(1)
3589         p = capture[0]
3590         try:
3591             ip = p[IP]
3592             tcp = p[TCP]
3593             self.assertEqual(ip.dst, self.pg1.remote_ip4)
3594             self.assertEqual(ip.src, self.nat_addr)
3595             self.assertEqual(tcp.dport, 22)
3596             self.assertNotEqual(tcp.sport, 4567)
3597             self.assertEqual((tcp.sport >> 6) & 63, 10)
3598             self.check_tcp_checksum(p)
3599             self.check_ip_checksum(p)
3600         except:
3601             self.logger.error(ppp("Unexpected or invalid packet:", p))
3602             raise
3603
3604     def test_twice_nat(self):
3605         """ Twice NAT44 """
3606         twice_nat_addr = '10.0.1.3'
3607         port_in = 8080
3608         port_out = 80
3609         eh_port_out = 4567
3610         eh_port_in = 0
3611         self.nat44_add_address(self.nat_addr)
3612         self.nat44_add_address(twice_nat_addr, twice_nat=1)
3613         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
3614                                       port_in, port_out, proto=IP_PROTOS.tcp,
3615                                       twice_nat=1)
3616         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3617         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3618                                                   is_inside=0)
3619
3620         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3621              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3622              TCP(sport=eh_port_out, dport=port_out))
3623         self.pg1.add_stream(p)
3624         self.pg_enable_capture(self.pg_interfaces)
3625         self.pg_start()
3626         capture = self.pg0.get_capture(1)
3627         p = capture[0]
3628         try:
3629             ip = p[IP]
3630             tcp = p[TCP]
3631             self.assertEqual(ip.dst, self.pg0.remote_ip4)
3632             self.assertEqual(ip.src, twice_nat_addr)
3633             self.assertEqual(tcp.dport, port_in)
3634             self.assertNotEqual(tcp.sport, eh_port_out)
3635             eh_port_in = tcp.sport
3636             self.check_tcp_checksum(p)
3637             self.check_ip_checksum(p)
3638         except:
3639             self.logger.error(ppp("Unexpected or invalid packet:", p))
3640             raise
3641
3642         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3643              IP(src=self.pg0.remote_ip4, dst=twice_nat_addr) /
3644              TCP(sport=port_in, dport=eh_port_in))
3645         self.pg0.add_stream(p)
3646         self.pg_enable_capture(self.pg_interfaces)
3647         self.pg_start()
3648         capture = self.pg1.get_capture(1)
3649         p = capture[0]
3650         try:
3651             ip = p[IP]
3652             tcp = p[TCP]
3653             self.assertEqual(ip.dst, self.pg1.remote_ip4)
3654             self.assertEqual(ip.src, self.nat_addr)
3655             self.assertEqual(tcp.dport, eh_port_out)
3656             self.assertEqual(tcp.sport, port_out)
3657             self.check_tcp_checksum(p)
3658             self.check_ip_checksum(p)
3659         except:
3660             self.logger.error(ppp("Unexpected or invalid packet:", p))
3661             raise
3662
3663     def test_twice_nat_lb(self):
3664         """ Twice NAT44 local service load balancing """
3665         external_addr_n = socket.inet_pton(socket.AF_INET, self.nat_addr)
3666         twice_nat_addr = '10.0.1.3'
3667         local_port = 8080
3668         external_port = 80
3669         eh_port_out = 4567
3670         eh_port_in = 0
3671         server1 = self.pg0.remote_hosts[0]
3672         server2 = self.pg0.remote_hosts[1]
3673
3674         locals = [{'addr': server1.ip4n,
3675                    'port': local_port,
3676                    'probability': 50},
3677                   {'addr': server2.ip4n,
3678                    'port': local_port,
3679                    'probability': 50}]
3680
3681         self.nat44_add_address(self.nat_addr)
3682         self.nat44_add_address(twice_nat_addr, twice_nat=1)
3683
3684         self.vapi.nat44_add_del_lb_static_mapping(external_addr_n,
3685                                                   external_port,
3686                                                   IP_PROTOS.tcp,
3687                                                   twice_nat=1,
3688                                                   local_num=len(locals),
3689                                                   locals=locals)
3690         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3691         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3692                                                   is_inside=0)
3693
3694         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
3695              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3696              TCP(sport=eh_port_out, dport=external_port))
3697         self.pg1.add_stream(p)
3698         self.pg_enable_capture(self.pg_interfaces)
3699         self.pg_start()
3700         capture = self.pg0.get_capture(1)
3701         p = capture[0]
3702         server = None
3703         try:
3704             ip = p[IP]
3705             tcp = p[TCP]
3706             self.assertEqual(ip.src, twice_nat_addr)
3707             self.assertIn(ip.dst, [server1.ip4, server2.ip4])
3708             if ip.dst == server1.ip4:
3709                 server = server1
3710             else:
3711                 server = server2
3712             self.assertNotEqual(tcp.sport, eh_port_out)
3713             eh_port_in = tcp.sport
3714             self.assertEqual(tcp.dport, local_port)
3715             self.check_tcp_checksum(p)
3716             self.check_ip_checksum(p)
3717         except:
3718             self.logger.error(ppp("Unexpected or invalid packet:", p))
3719             raise
3720
3721         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
3722              IP(src=server.ip4, dst=twice_nat_addr) /
3723              TCP(sport=local_port, dport=eh_port_in))
3724         self.pg0.add_stream(p)
3725         self.pg_enable_capture(self.pg_interfaces)
3726         self.pg_start()
3727         capture = self.pg1.get_capture(1)
3728         p = capture[0]
3729         try:
3730             ip = p[IP]
3731             tcp = p[TCP]
3732             self.assertEqual(ip.src, self.nat_addr)
3733             self.assertEqual(ip.dst, self.pg1.remote_ip4)
3734             self.assertEqual(tcp.sport, external_port)
3735             self.assertEqual(tcp.dport, eh_port_out)
3736             self.check_tcp_checksum(p)
3737             self.check_ip_checksum(p)
3738         except:
3739             self.logger.error(ppp("Unexpected or invalid packet:", p))
3740             raise
3741
3742     def test_twice_nat_interface_addr(self):
3743         """ Acquire twice NAT44 addresses from interface """
3744         self.vapi.nat44_add_interface_addr(self.pg7.sw_if_index, twice_nat=1)
3745
3746         # no address in NAT pool
3747         adresses = self.vapi.nat44_address_dump()
3748         self.assertEqual(0, len(adresses))
3749
3750         # configure interface address and check NAT address pool
3751         self.pg7.config_ip4()
3752         adresses = self.vapi.nat44_address_dump()
3753         self.assertEqual(1, len(adresses))
3754         self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
3755         self.assertEqual(adresses[0].twice_nat, 1)
3756
3757         # remove interface address and check NAT address pool
3758         self.pg7.unconfig_ip4()
3759         adresses = self.vapi.nat44_address_dump()
3760         self.assertEqual(0, len(adresses))
3761
3762     def test_ipfix_max_frags(self):
3763         """ IPFIX logging maximum fragments pending reassembly exceeded """
3764         self.nat44_add_address(self.nat_addr)
3765         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3766         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
3767                                                   is_inside=0)
3768         self.vapi.nat_set_reass(max_frag=0)
3769         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
3770                                      src_address=self.pg3.local_ip4n,
3771                                      path_mtu=512,
3772                                      template_interval=10)
3773         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
3774                             src_port=self.ipfix_src_port)
3775
3776         data = "A" * 4 + "B" * 16 + "C" * 3
3777         self.tcp_port_in = random.randint(1025, 65535)
3778         pkts = self.create_stream_frag(self.pg0,
3779                                        self.pg1.remote_ip4,
3780                                        self.tcp_port_in,
3781                                        20,
3782                                        data)
3783         self.pg0.add_stream(pkts[-1])
3784         self.pg_enable_capture(self.pg_interfaces)
3785         self.pg_start()
3786         frags = self.pg1.get_capture(0)
3787         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
3788         capture = self.pg3.get_capture(9)
3789         ipfix = IPFIXDecoder()
3790         # first load template
3791         for p in capture:
3792             self.assertTrue(p.haslayer(IPFIX))
3793             self.assertEqual(p[IP].src, self.pg3.local_ip4)
3794             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
3795             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
3796             self.assertEqual(p[UDP].dport, 4739)
3797             self.assertEqual(p[IPFIX].observationDomainID,
3798                              self.ipfix_domain_id)
3799             if p.haslayer(Template):
3800                 ipfix.add_template(p.getlayer(Template))
3801         # verify events in data set
3802         for p in capture:
3803             if p.haslayer(Data):
3804                 data = ipfix.decode_data_set(p.getlayer(Set))
3805                 self.verify_ipfix_max_fragments_ip4(data, 0,
3806                                                     self.pg0.remote_ip4n)
3807
3808     def tearDown(self):
3809         super(TestNAT44, self).tearDown()
3810         if not self.vpp_dead:
3811             self.logger.info(self.vapi.cli("show nat44 addresses"))
3812             self.logger.info(self.vapi.cli("show nat44 interfaces"))
3813             self.logger.info(self.vapi.cli("show nat44 static mappings"))
3814             self.logger.info(self.vapi.cli("show nat44 interface address"))
3815             self.logger.info(self.vapi.cli("show nat44 sessions detail"))
3816             self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
3817             self.vapi.cli("nat addr-port-assignment-alg default")
3818             self.clear_nat44()
3819
3820
3821 class TestNAT44Out2InDPO(MethodHolder):
3822     """ NAT44 Test Cases using out2in DPO """
3823
3824     @classmethod
3825     def setUpConstants(cls):
3826         super(TestNAT44Out2InDPO, cls).setUpConstants()
3827         cls.vpp_cmdline.extend(["nat", "{", "out2in dpo", "}"])
3828
3829     @classmethod
3830     def setUpClass(cls):
3831         super(TestNAT44Out2InDPO, cls).setUpClass()
3832
3833         try:
3834             cls.tcp_port_in = 6303
3835             cls.tcp_port_out = 6303
3836             cls.udp_port_in = 6304
3837             cls.udp_port_out = 6304
3838             cls.icmp_id_in = 6305
3839             cls.icmp_id_out = 6305
3840             cls.nat_addr = '10.0.0.3'
3841             cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
3842             cls.dst_ip4 = '192.168.70.1'
3843
3844             cls.create_pg_interfaces(range(2))
3845
3846             cls.pg0.admin_up()
3847             cls.pg0.config_ip4()
3848             cls.pg0.resolve_arp()
3849
3850             cls.pg1.admin_up()
3851             cls.pg1.config_ip6()
3852             cls.pg1.resolve_ndp()
3853
3854             cls.vapi.ip_add_del_route(is_ipv6=True, dst_address='\x00'*16,
3855                                       dst_address_length=0,
3856                                       next_hop_address=cls.pg1.remote_ip6n,
3857                                       next_hop_sw_if_index=cls.pg1.sw_if_index)
3858
3859         except Exception:
3860             super(TestNAT44Out2InDPO, cls).tearDownClass()
3861             raise
3862
3863     def configure_xlat(self):
3864         self.dst_ip6_pfx = '1:2:3::'
3865         self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3866                                               self.dst_ip6_pfx)
3867         self.dst_ip6_pfx_len = 96
3868         self.src_ip6_pfx = '4:5:6::'
3869         self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3870                                               self.src_ip6_pfx)
3871         self.src_ip6_pfx_len = 96
3872         self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
3873                                  self.src_ip6_pfx_n, self.src_ip6_pfx_len,
3874                                  '\x00\x00\x00\x00', 0, is_translation=1,
3875                                  is_rfc6052=1)
3876
3877     def test_464xlat_ce(self):
3878         """ Test 464XLAT CE with NAT44 """
3879
3880         self.configure_xlat()
3881
3882         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
3883         self.vapi.nat44_add_del_address_range(self.nat_addr_n, self.nat_addr_n)
3884
3885         out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3886                                        self.dst_ip6_pfx_len)
3887         out_dst_ip6 = self.compose_ip6(self.nat_addr, self.src_ip6_pfx,
3888                                        self.src_ip6_pfx_len)
3889
3890         try:
3891             pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3892             self.pg0.add_stream(pkts)
3893             self.pg_enable_capture(self.pg_interfaces)
3894             self.pg_start()
3895             capture = self.pg1.get_capture(len(pkts))
3896             self.verify_capture_out_ip6(capture, nat_ip=out_dst_ip6,
3897                                         dst_ip=out_src_ip6)
3898
3899             pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6,
3900                                               out_dst_ip6)
3901             self.pg1.add_stream(pkts)
3902             self.pg_enable_capture(self.pg_interfaces)
3903             self.pg_start()
3904             capture = self.pg0.get_capture(len(pkts))
3905             self.verify_capture_in(capture, self.pg0)
3906         finally:
3907             self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index,
3908                                                       is_add=0)
3909             self.vapi.nat44_add_del_address_range(self.nat_addr_n,
3910                                                   self.nat_addr_n, is_add=0)
3911
3912     def test_464xlat_ce_no_nat(self):
3913         """ Test 464XLAT CE without NAT44 """
3914
3915         self.configure_xlat()
3916
3917         out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3918                                        self.dst_ip6_pfx_len)
3919         out_dst_ip6 = self.compose_ip6(self.pg0.remote_ip4, self.src_ip6_pfx,
3920                                        self.src_ip6_pfx_len)
3921
3922         pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3923         self.pg0.add_stream(pkts)
3924         self.pg_enable_capture(self.pg_interfaces)
3925         self.pg_start()
3926         capture = self.pg1.get_capture(len(pkts))
3927         self.verify_capture_out_ip6(capture, dst_ip=out_src_ip6,
3928                                     nat_ip=out_dst_ip6, same_port=True)
3929
3930         pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6, out_dst_ip6)
3931         self.pg1.add_stream(pkts)
3932         self.pg_enable_capture(self.pg_interfaces)
3933         self.pg_start()
3934         capture = self.pg0.get_capture(len(pkts))
3935         self.verify_capture_in(capture, self.pg0)
3936
3937
3938 class TestDeterministicNAT(MethodHolder):
3939     """ Deterministic NAT Test Cases """
3940
3941     @classmethod
3942     def setUpConstants(cls):
3943         super(TestDeterministicNAT, cls).setUpConstants()
3944         cls.vpp_cmdline.extend(["nat", "{", "deterministic", "}"])
3945
3946     @classmethod
3947     def setUpClass(cls):
3948         super(TestDeterministicNAT, cls).setUpClass()
3949
3950         try:
3951             cls.tcp_port_in = 6303
3952             cls.tcp_external_port = 6303
3953             cls.udp_port_in = 6304
3954             cls.udp_external_port = 6304
3955             cls.icmp_id_in = 6305
3956             cls.nat_addr = '10.0.0.3'
3957
3958             cls.create_pg_interfaces(range(3))
3959             cls.interfaces = list(cls.pg_interfaces)
3960
3961             for i in cls.interfaces:
3962                 i.admin_up()
3963                 i.config_ip4()
3964                 i.resolve_arp()
3965
3966             cls.pg0.generate_remote_hosts(2)
3967             cls.pg0.configure_ipv4_neighbors()
3968
3969         except Exception:
3970             super(TestDeterministicNAT, cls).tearDownClass()
3971             raise
3972
3973     def create_stream_in(self, in_if, out_if, ttl=64):
3974         """
3975         Create packet stream for inside network
3976
3977         :param in_if: Inside interface
3978         :param out_if: Outside interface
3979         :param ttl: TTL of generated packets
3980         """
3981         pkts = []
3982         # TCP
3983         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
3984              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
3985              TCP(sport=self.tcp_port_in, dport=self.tcp_external_port))
3986         pkts.append(p)
3987
3988         # UDP
3989         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
3990              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
3991              UDP(sport=self.udp_port_in, dport=self.udp_external_port))
3992         pkts.append(p)
3993
3994         # ICMP
3995         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
3996              IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
3997              ICMP(id=self.icmp_id_in, type='echo-request'))
3998         pkts.append(p)
3999
4000         return pkts
4001
4002     def create_stream_out(self, out_if, dst_ip=None, ttl=64):
4003         """
4004         Create packet stream for outside network
4005
4006         :param out_if: Outside interface
4007         :param dst_ip: Destination IP address (Default use global NAT address)
4008         :param ttl: TTL of generated packets
4009         """
4010         if dst_ip is None:
4011             dst_ip = self.nat_addr
4012         pkts = []
4013         # TCP
4014         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
4015              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
4016              TCP(dport=self.tcp_port_out, sport=self.tcp_external_port))
4017         pkts.append(p)
4018
4019         # UDP
4020         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
4021              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
4022              UDP(dport=self.udp_port_out, sport=self.udp_external_port))
4023         pkts.append(p)
4024
4025         # ICMP
4026         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
4027              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
4028              ICMP(id=self.icmp_external_id, type='echo-reply'))
4029         pkts.append(p)
4030
4031         return pkts
4032
4033     def verify_capture_out(self, capture, nat_ip=None, packet_num=3):
4034         """
4035         Verify captured packets on outside network
4036
4037         :param capture: Captured packets
4038         :param nat_ip: Translated IP address (Default use global NAT address)
4039         :param same_port: Sorce port number is not translated (Default False)
4040         :param packet_num: Expected number of packets (Default 3)
4041         """
4042         if nat_ip is None:
4043             nat_ip = self.nat_addr
4044         self.assertEqual(packet_num, len(capture))
4045         for packet in capture:
4046             try:
4047                 self.assertEqual(packet[IP].src, nat_ip)
4048                 if packet.haslayer(TCP):
4049                     self.tcp_port_out = packet[TCP].sport
4050                 elif packet.haslayer(UDP):
4051                     self.udp_port_out = packet[UDP].sport
4052                 else:
4053                     self.icmp_external_id = packet[ICMP].id
4054             except:
4055                 self.logger.error(ppp("Unexpected or invalid packet "
4056                                       "(outside network):", packet))
4057                 raise
4058
4059     def initiate_tcp_session(self, in_if, out_if):
4060         """
4061         Initiates TCP session
4062
4063         :param in_if: Inside interface
4064         :param out_if: Outside interface
4065         """
4066         try:
4067             # SYN packet in->out
4068             p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
4069                  IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
4070                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4071                      flags="S"))
4072             in_if.add_stream(p)
4073             self.pg_enable_capture(self.pg_interfaces)
4074             self.pg_start()
4075             capture = out_if.get_capture(1)
4076             p = capture[0]
4077             self.tcp_port_out = p[TCP].sport
4078
4079             # SYN + ACK packet out->in
4080             p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
4081                  IP(src=out_if.remote_ip4, dst=self.nat_addr) /
4082                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
4083                      flags="SA"))
4084             out_if.add_stream(p)
4085             self.pg_enable_capture(self.pg_interfaces)
4086             self.pg_start()
4087             in_if.get_capture(1)
4088
4089             # ACK packet in->out
4090             p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
4091                  IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
4092                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4093                      flags="A"))
4094             in_if.add_stream(p)
4095             self.pg_enable_capture(self.pg_interfaces)
4096             self.pg_start()
4097             out_if.get_capture(1)
4098
4099         except:
4100             self.logger.error("TCP 3 way handshake failed")
4101             raise
4102
4103     def verify_ipfix_max_entries_per_user(self, data):
4104         """
4105         Verify IPFIX maximum entries per user exceeded event
4106
4107         :param data: Decoded IPFIX data records
4108         """
4109         self.assertEqual(1, len(data))
4110         record = data[0]
4111         # natEvent
4112         self.assertEqual(ord(record[230]), 13)
4113         # natQuotaExceededEvent
4114         self.assertEqual('\x03\x00\x00\x00', record[466])
4115         # maxEntriesPerUser
4116         self.assertEqual('\xe8\x03\x00\x00', record[473])
4117         # sourceIPv4Address
4118         self.assertEqual(self.pg0.remote_ip4n, record[8])
4119
4120     def test_deterministic_mode(self):
4121         """ NAT plugin run deterministic mode """
4122         in_addr = '172.16.255.0'
4123         out_addr = '172.17.255.50'
4124         in_addr_t = '172.16.255.20'
4125         in_addr_n = socket.inet_aton(in_addr)
4126         out_addr_n = socket.inet_aton(out_addr)
4127         in_addr_t_n = socket.inet_aton(in_addr_t)
4128         in_plen = 24
4129         out_plen = 32
4130
4131         nat_config = self.vapi.nat_show_config()
4132         self.assertEqual(1, nat_config.deterministic)
4133
4134         self.vapi.nat_det_add_del_map(in_addr_n, in_plen, out_addr_n, out_plen)
4135
4136         rep1 = self.vapi.nat_det_forward(in_addr_t_n)
4137         self.assertEqual(rep1.out_addr[:4], out_addr_n)
4138         rep2 = self.vapi.nat_det_reverse(out_addr_n, rep1.out_port_hi)
4139         self.assertEqual(rep2.in_addr[:4], in_addr_t_n)
4140
4141         deterministic_mappings = self.vapi.nat_det_map_dump()
4142         self.assertEqual(len(deterministic_mappings), 1)
4143         dsm = deterministic_mappings[0]
4144         self.assertEqual(in_addr_n, dsm.in_addr[:4])
4145         self.assertEqual(in_plen, dsm.in_plen)
4146         self.assertEqual(out_addr_n, dsm.out_addr[:4])
4147         self.assertEqual(out_plen, dsm.out_plen)
4148
4149         self.clear_nat_det()
4150         deterministic_mappings = self.vapi.nat_det_map_dump()
4151         self.assertEqual(len(deterministic_mappings), 0)
4152
4153     def test_set_timeouts(self):
4154         """ Set deterministic NAT timeouts """
4155         timeouts_before = self.vapi.nat_det_get_timeouts()
4156
4157         self.vapi.nat_det_set_timeouts(timeouts_before.udp + 10,
4158                                        timeouts_before.tcp_established + 10,
4159                                        timeouts_before.tcp_transitory + 10,
4160                                        timeouts_before.icmp + 10)
4161
4162         timeouts_after = self.vapi.nat_det_get_timeouts()
4163
4164         self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
4165         self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
4166         self.assertNotEqual(timeouts_before.tcp_established,
4167                             timeouts_after.tcp_established)
4168         self.assertNotEqual(timeouts_before.tcp_transitory,
4169                             timeouts_after.tcp_transitory)
4170
4171     def test_det_in(self):
4172         """ Deterministic NAT translation test (TCP, UDP, ICMP) """
4173
4174         nat_ip = "10.0.0.10"
4175
4176         self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
4177                                       32,
4178                                       socket.inet_aton(nat_ip),
4179                                       32)
4180         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4181         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4182                                                   is_inside=0)
4183
4184         # in2out
4185         pkts = self.create_stream_in(self.pg0, self.pg1)
4186         self.pg0.add_stream(pkts)
4187         self.pg_enable_capture(self.pg_interfaces)
4188         self.pg_start()
4189         capture = self.pg1.get_capture(len(pkts))
4190         self.verify_capture_out(capture, nat_ip)
4191
4192         # out2in
4193         pkts = self.create_stream_out(self.pg1, nat_ip)
4194         self.pg1.add_stream(pkts)
4195         self.pg_enable_capture(self.pg_interfaces)
4196         self.pg_start()
4197         capture = self.pg0.get_capture(len(pkts))
4198         self.verify_capture_in(capture, self.pg0)
4199
4200         # session dump test
4201         sessions = self.vapi.nat_det_session_dump(self.pg0.remote_ip4n)
4202         self.assertEqual(len(sessions), 3)
4203
4204         # TCP session
4205         s = sessions[0]
4206         self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
4207         self.assertEqual(s.in_port, self.tcp_port_in)
4208         self.assertEqual(s.out_port, self.tcp_port_out)
4209         self.assertEqual(s.ext_port, self.tcp_external_port)
4210
4211         # UDP session
4212         s = sessions[1]
4213         self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
4214         self.assertEqual(s.in_port, self.udp_port_in)
4215         self.assertEqual(s.out_port, self.udp_port_out)
4216         self.assertEqual(s.ext_port, self.udp_external_port)
4217
4218         # ICMP session
4219         s = sessions[2]
4220         self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
4221         self.assertEqual(s.in_port, self.icmp_id_in)
4222         self.assertEqual(s.out_port, self.icmp_external_id)
4223
4224     def test_multiple_users(self):
4225         """ Deterministic NAT multiple users """
4226
4227         nat_ip = "10.0.0.10"
4228         port_in = 80
4229         external_port = 6303
4230
4231         host0 = self.pg0.remote_hosts[0]
4232         host1 = self.pg0.remote_hosts[1]
4233
4234         self.vapi.nat_det_add_del_map(host0.ip4n,
4235                                       24,
4236                                       socket.inet_aton(nat_ip),
4237                                       32)
4238         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4239         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4240                                                   is_inside=0)
4241
4242         # host0 to out
4243         p = (Ether(src=host0.mac, dst=self.pg0.local_mac) /
4244              IP(src=host0.ip4, dst=self.pg1.remote_ip4) /
4245              TCP(sport=port_in, dport=external_port))
4246         self.pg0.add_stream(p)
4247         self.pg_enable_capture(self.pg_interfaces)
4248         self.pg_start()
4249         capture = self.pg1.get_capture(1)
4250         p = capture[0]
4251         try:
4252             ip = p[IP]
4253             tcp = p[TCP]
4254             self.assertEqual(ip.src, nat_ip)
4255             self.assertEqual(ip.dst, self.pg1.remote_ip4)
4256             self.assertEqual(tcp.dport, external_port)
4257             port_out0 = tcp.sport
4258         except:
4259             self.logger.error(ppp("Unexpected or invalid packet:", p))
4260             raise
4261
4262         # host1 to out
4263         p = (Ether(src=host1.mac, dst=self.pg0.local_mac) /
4264              IP(src=host1.ip4, dst=self.pg1.remote_ip4) /
4265              TCP(sport=port_in, dport=external_port))
4266         self.pg0.add_stream(p)
4267         self.pg_enable_capture(self.pg_interfaces)
4268         self.pg_start()
4269         capture = self.pg1.get_capture(1)
4270         p = capture[0]
4271         try:
4272             ip = p[IP]
4273             tcp = p[TCP]
4274             self.assertEqual(ip.src, nat_ip)
4275             self.assertEqual(ip.dst, self.pg1.remote_ip4)
4276             self.assertEqual(tcp.dport, external_port)
4277             port_out1 = tcp.sport
4278         except:
4279             self.logger.error(ppp("Unexpected or invalid packet:", p))
4280             raise
4281
4282         dms = self.vapi.nat_det_map_dump()
4283         self.assertEqual(1, len(dms))
4284         self.assertEqual(2, dms[0].ses_num)
4285
4286         # out to host0
4287         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4288              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
4289              TCP(sport=external_port, dport=port_out0))
4290         self.pg1.add_stream(p)
4291         self.pg_enable_capture(self.pg_interfaces)
4292         self.pg_start()
4293         capture = self.pg0.get_capture(1)
4294         p = capture[0]
4295         try:
4296             ip = p[IP]
4297             tcp = p[TCP]
4298             self.assertEqual(ip.src, self.pg1.remote_ip4)
4299             self.assertEqual(ip.dst, host0.ip4)
4300             self.assertEqual(tcp.dport, port_in)
4301             self.assertEqual(tcp.sport, external_port)
4302         except:
4303             self.logger.error(ppp("Unexpected or invalid packet:", p))
4304             raise
4305
4306         # out to host1
4307         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4308              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
4309              TCP(sport=external_port, dport=port_out1))
4310         self.pg1.add_stream(p)
4311         self.pg_enable_capture(self.pg_interfaces)
4312         self.pg_start()
4313         capture = self.pg0.get_capture(1)
4314         p = capture[0]
4315         try:
4316             ip = p[IP]
4317             tcp = p[TCP]
4318             self.assertEqual(ip.src, self.pg1.remote_ip4)
4319             self.assertEqual(ip.dst, host1.ip4)
4320             self.assertEqual(tcp.dport, port_in)
4321             self.assertEqual(tcp.sport, external_port)
4322         except:
4323             self.logger.error(ppp("Unexpected or invalid packet", p))
4324             raise
4325
4326         # session close api test
4327         self.vapi.nat_det_close_session_out(socket.inet_aton(nat_ip),
4328                                             port_out1,
4329                                             self.pg1.remote_ip4n,
4330                                             external_port)
4331         dms = self.vapi.nat_det_map_dump()
4332         self.assertEqual(dms[0].ses_num, 1)
4333
4334         self.vapi.nat_det_close_session_in(host0.ip4n,
4335                                            port_in,
4336                                            self.pg1.remote_ip4n,
4337                                            external_port)
4338         dms = self.vapi.nat_det_map_dump()
4339         self.assertEqual(dms[0].ses_num, 0)
4340
4341     def test_tcp_session_close_detection_in(self):
4342         """ Deterministic NAT TCP session close from inside network """
4343         self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
4344                                       32,
4345                                       socket.inet_aton(self.nat_addr),
4346                                       32)
4347         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4348         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4349                                                   is_inside=0)
4350
4351         self.initiate_tcp_session(self.pg0, self.pg1)
4352
4353         # close the session from inside
4354         try:
4355             # FIN packet in -> out
4356             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4357                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4358                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4359                      flags="F"))
4360             self.pg0.add_stream(p)
4361             self.pg_enable_capture(self.pg_interfaces)
4362             self.pg_start()
4363             self.pg1.get_capture(1)
4364
4365             pkts = []
4366
4367             # ACK packet out -> in
4368             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4369                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
4370                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
4371                      flags="A"))
4372             pkts.append(p)
4373
4374             # FIN packet out -> in
4375             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4376                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
4377                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
4378                      flags="F"))
4379             pkts.append(p)
4380
4381             self.pg1.add_stream(pkts)
4382             self.pg_enable_capture(self.pg_interfaces)
4383             self.pg_start()
4384             self.pg0.get_capture(2)
4385
4386             # ACK packet in -> out
4387             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4388                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4389                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4390                      flags="A"))
4391             self.pg0.add_stream(p)
4392             self.pg_enable_capture(self.pg_interfaces)
4393             self.pg_start()
4394             self.pg1.get_capture(1)
4395
4396             # Check if deterministic NAT44 closed the session
4397             dms = self.vapi.nat_det_map_dump()
4398             self.assertEqual(0, dms[0].ses_num)
4399         except:
4400             self.logger.error("TCP session termination failed")
4401             raise
4402
4403     def test_tcp_session_close_detection_out(self):
4404         """ Deterministic NAT TCP session close from outside network """
4405         self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
4406                                       32,
4407                                       socket.inet_aton(self.nat_addr),
4408                                       32)
4409         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4410         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4411                                                   is_inside=0)
4412
4413         self.initiate_tcp_session(self.pg0, self.pg1)
4414
4415         # close the session from outside
4416         try:
4417             # FIN packet out -> in
4418             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4419                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
4420                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
4421                      flags="F"))
4422             self.pg1.add_stream(p)
4423             self.pg_enable_capture(self.pg_interfaces)
4424             self.pg_start()
4425             self.pg0.get_capture(1)
4426
4427             pkts = []
4428
4429             # ACK packet in -> out
4430             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4431                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4432                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4433                      flags="A"))
4434             pkts.append(p)
4435
4436             # ACK packet in -> out
4437             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4438                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4439                  TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
4440                      flags="F"))
4441             pkts.append(p)
4442
4443             self.pg0.add_stream(pkts)
4444             self.pg_enable_capture(self.pg_interfaces)
4445             self.pg_start()
4446             self.pg1.get_capture(2)
4447
4448             # ACK packet out -> in
4449             p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
4450                  IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
4451                  TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
4452                      flags="A"))
4453             self.pg1.add_stream(p)
4454             self.pg_enable_capture(self.pg_interfaces)
4455             self.pg_start()
4456             self.pg0.get_capture(1)
4457
4458             # Check if deterministic NAT44 closed the session
4459             dms = self.vapi.nat_det_map_dump()
4460             self.assertEqual(0, dms[0].ses_num)
4461         except:
4462             self.logger.error("TCP session termination failed")
4463             raise
4464
4465     @unittest.skipUnless(running_extended_tests(), "part of extended tests")
4466     def test_session_timeout(self):
4467         """ Deterministic NAT session timeouts """
4468         self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
4469                                       32,
4470                                       socket.inet_aton(self.nat_addr),
4471                                       32)
4472         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4473         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4474                                                   is_inside=0)
4475
4476         self.initiate_tcp_session(self.pg0, self.pg1)
4477         self.vapi.nat_det_set_timeouts(5, 5, 5, 5)
4478         pkts = self.create_stream_in(self.pg0, self.pg1)
4479         self.pg0.add_stream(pkts)
4480         self.pg_enable_capture(self.pg_interfaces)
4481         self.pg_start()
4482         capture = self.pg1.get_capture(len(pkts))
4483         sleep(15)
4484
4485         dms = self.vapi.nat_det_map_dump()
4486         self.assertEqual(0, dms[0].ses_num)
4487
4488     @unittest.skipUnless(running_extended_tests(), "part of extended tests")
4489     def test_session_limit_per_user(self):
4490         """ Deterministic NAT maximum sessions per user limit """
4491         self.vapi.nat_det_add_del_map(self.pg0.remote_ip4n,
4492                                       32,
4493                                       socket.inet_aton(self.nat_addr),
4494                                       32)
4495         self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
4496         self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
4497                                                   is_inside=0)
4498         self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n,
4499                                      src_address=self.pg2.local_ip4n,
4500                                      path_mtu=512,
4501                                      template_interval=10)
4502         self.vapi.nat_ipfix()
4503
4504         pkts = []
4505         for port in range(1025, 2025):
4506             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4507                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4508                  UDP(sport=port, dport=port))
4509             pkts.append(p)
4510
4511         self.pg0.add_stream(pkts)
4512         self.pg_enable_capture(self.pg_interfaces)
4513         self.pg_start()
4514         capture = self.pg1.get_capture(len(pkts))
4515
4516         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
4517              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
4518              UDP(sport=3001, dport=3002))
4519         self.pg0.add_stream(p)
4520         self.pg_enable_capture(self.pg_interfaces)
4521         self.pg_start()
4522         capture = self.pg1.assert_nothing_captured()
4523
4524         # verify ICMP error packet
4525         capture = self.pg0.get_capture(1)
4526         p = capture[0]
4527         self.assertTrue(p.haslayer(ICMP))
4528         icmp = p[ICMP]
4529         self.assertEqual(icmp.type, 3)
4530         self.assertEqual(icmp.code, 1)
4531         self.assertTrue(icmp.haslayer(IPerror))
4532         inner_ip = icmp[IPerror]
4533         self.assertEqual(inner_ip[UDPerror].sport, 3001)
4534         self.assertEqual(inner_ip[UDPerror].dport, 3002)
4535
4536         dms = self.vapi.nat_det_map_dump()
4537
4538         self.assertEqual(1000, dms[0].ses_num)
4539
4540         # verify IPFIX logging
4541         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
4542         sleep(1)
4543         capture = self.pg2.get_capture(2)
4544         ipfix = IPFIXDecoder()
4545         # first load template
4546         for p in capture:
4547             self.assertTrue(p.haslayer(IPFIX))
4548             if p.haslayer(Template):
4549                 ipfix.add_template(p.getlayer(Template))
4550         # verify events in data set
4551         for p in capture:
4552             if p.haslayer(Data):
4553                 data = ipfix.decode_data_set(p.getlayer(Set))
4554                 self.verify_ipfix_max_entries_per_user(data)
4555
4556     def clear_nat_det(self):
4557         """
4558         Clear deterministic NAT configuration.
4559         """
4560         self.vapi.nat_ipfix(enable=0)
4561         self.vapi.nat_det_set_timeouts()
4562         deterministic_mappings = self.vapi.nat_det_map_dump()
4563         for dsm in deterministic_mappings:
4564             self.vapi.nat_det_add_del_map(dsm.in_addr,
4565                                           dsm.in_plen,
4566                                           dsm.out_addr,
4567                                           dsm.out_plen,
4568                                           is_add=0)
4569
4570         interfaces = self.vapi.nat44_interface_dump()
4571         for intf in interfaces:
4572             self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
4573                                                       intf.is_inside,
4574                                                       is_add=0)
4575
4576     def tearDown(self):
4577         super(TestDeterministicNAT, self).tearDown()
4578         if not self.vpp_dead:
4579             self.logger.info(self.vapi.cli("show nat44 interfaces"))
4580             self.logger.info(
4581                 self.vapi.cli("show nat44 deterministic mappings"))
4582             self.logger.info(
4583                 self.vapi.cli("show nat44 deterministic timeouts"))
4584             self.logger.info(
4585                 self.vapi.cli("show nat44 deterministic sessions"))
4586             self.clear_nat_det()
4587
4588
4589 class TestNAT64(MethodHolder):
4590     """ NAT64 Test Cases """
4591
4592     @classmethod
4593     def setUpConstants(cls):
4594         super(TestNAT64, cls).setUpConstants()
4595         cls.vpp_cmdline.extend(["nat", "{", "nat64 bib hash buckets 128",
4596                                 "nat64 st hash buckets 256", "}"])
4597
4598     @classmethod
4599     def setUpClass(cls):
4600         super(TestNAT64, cls).setUpClass()
4601
4602         try:
4603             cls.tcp_port_in = 6303
4604             cls.tcp_port_out = 6303
4605             cls.udp_port_in = 6304
4606             cls.udp_port_out = 6304
4607             cls.icmp_id_in = 6305
4608             cls.icmp_id_out = 6305
4609             cls.nat_addr = '10.0.0.3'
4610             cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
4611             cls.vrf1_id = 10
4612             cls.vrf1_nat_addr = '10.0.10.3'
4613             cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
4614                                                    cls.vrf1_nat_addr)
4615             cls.ipfix_src_port = 4739
4616             cls.ipfix_domain_id = 1
4617
4618             cls.create_pg_interfaces(range(5))
4619             cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
4620             cls.ip6_interfaces.append(cls.pg_interfaces[2])
4621             cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
4622
4623             cls.vapi.ip_table_add_del(cls.vrf1_id, is_add=1, is_ipv6=1)
4624
4625             cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
4626
4627             cls.pg0.generate_remote_hosts(2)
4628
4629             for i in cls.ip6_interfaces:
4630                 i.admin_up()
4631                 i.config_ip6()
4632                 i.configure_ipv6_neighbors()
4633
4634             for i in cls.ip4_interfaces:
4635                 i.admin_up()
4636                 i.config_ip4()
4637                 i.resolve_arp()
4638
4639             cls.pg3.admin_up()
4640             cls.pg3.config_ip4()
4641             cls.pg3.resolve_arp()
4642             cls.pg3.config_ip6()
4643             cls.pg3.configure_ipv6_neighbors()
4644
4645         except Exception:
4646             super(TestNAT64, cls).tearDownClass()
4647             raise
4648
4649     def test_pool(self):
4650         """ Add/delete address to NAT64 pool """
4651         nat_addr = socket.inet_pton(socket.AF_INET, '1.2.3.4')
4652
4653         self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr)
4654
4655         addresses = self.vapi.nat64_pool_addr_dump()
4656         self.assertEqual(len(addresses), 1)
4657         self.assertEqual(addresses[0].address, nat_addr)
4658
4659         self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr, is_add=0)
4660
4661         addresses = self.vapi.nat64_pool_addr_dump()
4662         self.assertEqual(len(addresses), 0)
4663
4664     def test_interface(self):
4665         """ Enable/disable NAT64 feature on the interface """
4666         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
4667         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
4668
4669         interfaces = self.vapi.nat64_interface_dump()
4670         self.assertEqual(len(interfaces), 2)
4671         pg0_found = False
4672         pg1_found = False
4673         for intf in interfaces:
4674             if intf.sw_if_index == self.pg0.sw_if_index:
4675                 self.assertEqual(intf.is_inside, 1)
4676                 pg0_found = True
4677             elif intf.sw_if_index == self.pg1.sw_if_index:
4678                 self.assertEqual(intf.is_inside, 0)
4679                 pg1_found = True
4680         self.assertTrue(pg0_found)
4681         self.assertTrue(pg1_found)
4682
4683         features = self.vapi.cli("show interface features pg0")
4684         self.assertNotEqual(features.find('nat64-in2out'), -1)
4685         features = self.vapi.cli("show interface features pg1")
4686         self.assertNotEqual(features.find('nat64-out2in'), -1)
4687
4688         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index, is_add=0)
4689         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_add=0)
4690
4691         interfaces = self.vapi.nat64_interface_dump()
4692         self.assertEqual(len(interfaces), 0)
4693
4694     def test_static_bib(self):
4695         """ Add/delete static BIB entry """
4696         in_addr = socket.inet_pton(socket.AF_INET6,
4697                                    '2001:db8:85a3::8a2e:370:7334')
4698         out_addr = socket.inet_pton(socket.AF_INET, '10.1.1.3')
4699         in_port = 1234
4700         out_port = 5678
4701         proto = IP_PROTOS.tcp
4702
4703         self.vapi.nat64_add_del_static_bib(in_addr,
4704                                            out_addr,
4705                                            in_port,
4706                                            out_port,
4707                                            proto)
4708         bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
4709         static_bib_num = 0
4710         for bibe in bib:
4711             if bibe.is_static:
4712                 static_bib_num += 1
4713                 self.assertEqual(bibe.i_addr, in_addr)
4714                 self.assertEqual(bibe.o_addr, out_addr)
4715                 self.assertEqual(bibe.i_port, in_port)
4716                 self.assertEqual(bibe.o_port, out_port)
4717         self.assertEqual(static_bib_num, 1)
4718
4719         self.vapi.nat64_add_del_static_bib(in_addr,
4720                                            out_addr,
4721                                            in_port,
4722                                            out_port,
4723                                            proto,
4724                                            is_add=0)
4725         bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
4726         static_bib_num = 0
4727         for bibe in bib:
4728             if bibe.is_static:
4729                 static_bib_num += 1
4730         self.assertEqual(static_bib_num, 0)
4731
4732     def test_set_timeouts(self):
4733         """ Set NAT64 timeouts """
4734         # verify default values
4735         timeouts = self.vapi.nat64_get_timeouts()
4736         self.assertEqual(timeouts.udp, 300)
4737         self.assertEqual(timeouts.icmp, 60)
4738         self.assertEqual(timeouts.tcp_trans, 240)
4739         self.assertEqual(timeouts.tcp_est, 7440)
4740         self.assertEqual(timeouts.tcp_incoming_syn, 6)
4741
4742         # set and verify custom values
4743         self.vapi.nat64_set_timeouts(udp=200, icmp=30, tcp_trans=250,
4744                                      tcp_est=7450, tcp_incoming_syn=10)
4745         timeouts = self.vapi.nat64_get_timeouts()
4746         self.assertEqual(timeouts.udp, 200)
4747         self.assertEqual(timeouts.icmp, 30)
4748         self.assertEqual(timeouts.tcp_trans, 250)
4749         self.assertEqual(timeouts.tcp_est, 7450)
4750         self.assertEqual(timeouts.tcp_incoming_syn, 10)
4751
4752     def test_dynamic(self):
4753         """ NAT64 dynamic translation test """
4754         self.tcp_port_in = 6303
4755         self.udp_port_in = 6304
4756         self.icmp_id_in = 6305
4757
4758         ses_num_start = self.nat64_get_ses_num()
4759
4760         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
4761                                                 self.nat_addr_n)
4762         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
4763         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
4764
4765         # in2out
4766         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
4767         self.pg0.add_stream(pkts)
4768         self.pg_enable_capture(self.pg_interfaces)
4769         self.pg_start()
4770         capture = self.pg1.get_capture(len(pkts))
4771         self.verify_capture_out(capture, nat_ip=self.nat_addr,
4772                                 dst_ip=self.pg1.remote_ip4)
4773
4774         # out2in
4775         pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
4776         self.pg1.add_stream(pkts)
4777         self.pg_enable_capture(self.pg_interfaces)
4778         self.pg_start()
4779         capture = self.pg0.get_capture(len(pkts))
4780         ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
4781         self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
4782
4783         # in2out
4784         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
4785         self.pg0.add_stream(pkts)
4786         self.pg_enable_capture(self.pg_interfaces)
4787         self.pg_start()
4788         capture = self.pg1.get_capture(len(pkts))
4789         self.verify_capture_out(capture, nat_ip=self.nat_addr,
4790                                 dst_ip=self.pg1.remote_ip4)
4791
4792         # out2in
4793         pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
4794         self.pg1.add_stream(pkts)
4795         self.pg_enable_capture(self.pg_interfaces)
4796         self.pg_start()
4797         capture = self.pg0.get_capture(len(pkts))
4798         self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
4799
4800         ses_num_end = self.nat64_get_ses_num()
4801
4802         self.assertEqual(ses_num_end - ses_num_start, 3)
4803
4804         # tenant with specific VRF
4805         self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
4806                                                 self.vrf1_nat_addr_n,
4807                                                 vrf_id=self.vrf1_id)
4808         self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
4809
4810         pkts = self.create_stream_in_ip6(self.pg2, self.pg1)
4811         self.pg2.add_stream(pkts)
4812         self.pg_enable_capture(self.pg_interfaces)
4813         self.pg_start()
4814         capture = self.pg1.get_capture(len(pkts))
4815         self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
4816                                 dst_ip=self.pg1.remote_ip4)
4817
4818         pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
4819         self.pg1.add_stream(pkts)
4820         self.pg_enable_capture(self.pg_interfaces)
4821         self.pg_start()
4822         capture = self.pg2.get_capture(len(pkts))
4823         self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg2.remote_ip6)
4824
4825     def test_static(self):
4826         """ NAT64 static translation test """
4827         self.tcp_port_in = 60303
4828         self.udp_port_in = 60304
4829         self.icmp_id_in = 60305
4830         self.tcp_port_out = 60303
4831         self.udp_port_out = 60304
4832         self.icmp_id_out = 60305
4833
4834         ses_num_start = self.nat64_get_ses_num()
4835
4836         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
4837                                                 self.nat_addr_n)
4838         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
4839         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
4840
4841         self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
4842                                            self.nat_addr_n,
4843                                            self.tcp_port_in,
4844                                            self.tcp_port_out,
4845                                            IP_PROTOS.tcp)
4846         self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
4847                                            self.nat_addr_n,
4848                                            self.udp_port_in,
4849                                            self.udp_port_out,
4850                                            IP_PROTOS.udp)
4851         self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
4852                                            self.nat_addr_n,
4853                                            self.icmp_id_in,
4854                                            self.icmp_id_out,
4855                                            IP_PROTOS.icmp)
4856
4857         # in2out
4858         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
4859         self.pg0.add_stream(pkts)
4860         self.pg_enable_capture(self.pg_interfaces)
4861         self.pg_start()
4862         capture = self.pg1.get_capture(len(pkts))
4863         self.verify_capture_out(capture, nat_ip=self.nat_addr,
4864                                 dst_ip=self.pg1.remote_ip4, same_port=True)
4865
4866         # out2in
4867         pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
4868         self.pg1.add_stream(pkts)
4869         self.pg_enable_capture(self.pg_interfaces)
4870         self.pg_start()
4871         capture = self.pg0.get_capture(len(pkts))
4872         ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
4873         self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
4874
4875         ses_num_end = self.nat64_get_ses_num()
4876
4877         self.assertEqual(ses_num_end - ses_num_start, 3)
4878
4879     @unittest.skipUnless(running_extended_tests(), "part of extended tests")
4880     def test_session_timeout(self):
4881         """ NAT64 session timeout """
4882         self.icmp_id_in = 1234
4883         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
4884                                                 self.nat_addr_n)
4885         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
4886         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
4887         self.vapi.nat64_set_timeouts(icmp=5)
4888
4889         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
4890         self.pg0.add_stream(pkts)
4891         self.pg_enable_capture(self.pg_interfaces)
4892         self.pg_start()
4893         capture = self.pg1.get_capture(len(pkts))
4894
4895         ses_num_before_timeout = self.nat64_get_ses_num()
4896
4897         sleep(15)
4898
4899         # ICMP session after timeout
4900         ses_num_after_timeout = self.nat64_get_ses_num()
4901         self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout)
4902
4903     def test_icmp_error(self):
4904         """ NAT64 ICMP Error message translation """
4905         self.tcp_port_in = 6303
4906         self.udp_port_in = 6304
4907         self.icmp_id_in = 6305
4908
4909         ses_num_start = self.nat64_get_ses_num()
4910
4911         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
4912                                                 self.nat_addr_n)
4913         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
4914         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
4915
4916         # send some packets to create sessions
4917         pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
4918         self.pg0.add_stream(pkts)
4919         self.pg_enable_capture(self.pg_interfaces)
4920         self.pg_start()
4921         capture_ip4 = self.pg1.get_capture(len(pkts))
4922         self.verify_capture_out(capture_ip4,
4923                                 nat_ip=self.nat_addr,
4924                                 dst_ip=self.pg1.remote_ip4)
4925
4926         pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
4927         self.pg1.add_stream(pkts)
4928         self.pg_enable_capture(self.pg_interfaces)
4929         self.pg_start()
4930         capture_ip6 = self.pg0.get_capture(len(pkts))
4931         ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
4932         self.verify_capture_in_ip6(capture_ip6, ip[IPv6].src,
4933                                    self.pg0.remote_ip6)
4934
4935         # in2out
4936         pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4937                 IPv6(src=self.pg0.remote_ip6, dst=ip[IPv6].src) /
4938                 ICMPv6DestUnreach(code=1) /
4939                 packet[IPv6] for packet in capture_ip6]
4940         self.pg0.add_stream(pkts)
4941         self.pg_enable_capture(self.pg_interfaces)
4942         self.pg_start()
4943         capture = self.pg1.get_capture(len(pkts))
4944         for packet in capture:
4945             try:
4946                 self.assertEqual(packet[IP].src, self.nat_addr)
4947                 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
4948                 self.assertEqual(packet[ICMP].type, 3)
4949                 self.assertEqual(packet[ICMP].code, 13)
4950                 inner = packet[IPerror]
4951                 self.assertEqual(inner.src, self.pg1.remote_ip4)
4952                 self.assertEqual(inner.dst, self.nat_addr)
4953                 self.check_icmp_checksum(packet)
4954                 if inner.haslayer(TCPerror):
4955                     self.assertEqual(inner[TCPerror].dport, self.tcp_port_out)
4956                 elif inner.haslayer(UDPerror):
4957                     self.assertEqual(inner[UDPerror].dport, self.udp_port_out)
4958                 else:
4959                     self.assertEqual(inner[ICMPerror].id, self.icmp_id_out)
4960             except:
4961                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4962                 raise
4963
4964         # out2in
4965         pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
4966                 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
4967                 ICMP(type=3, code=13) /
4968                 packet[IP] for packet in capture_ip4]
4969         self.pg1.add_stream(pkts)
4970         self.pg_enable_capture(self.pg_interfaces)
4971         self.pg_start()
4972         capture = self.pg0.get_capture(len(pkts))
4973         for packet in capture:
4974             try:
4975                 self.assertEqual(packet[IPv6].src, ip.src)
4976                 self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
4977                 icmp = packet[ICMPv6DestUnreach]
4978                 self.assertEqual(icmp.code, 1)
4979                 inner = icmp[IPerror6]
4980                 self.assertEqual(inner.src, self.pg0.remote_ip6)
4981                 self.assertEqual(inner.dst, ip.src)
4982                 self.check_icmpv6_checksum(packet)
4983                 if inner.haslayer(TCPerror):
4984                     self.assertEqual(inner[TCPerror].sport, self.tcp_port_in)
4985                 elif inner.haslayer(UDPerror):
4986                     self.assertEqual(inner[UDPerror].sport, self.udp_port_in)
4987                 else:
4988                     self.assertEqual(inner[ICMPv6EchoRequest].id,
4989                                      self.icmp_id_in)
4990             except:
4991                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4992                 raise
4993
4994     def test_hairpinning(self):
4995         """ NAT64 hairpinning """
4996
4997         client = self.pg0.remote_hosts[0]
4998         server = self.pg0.remote_hosts[1]
4999         server_tcp_in_port = 22
5000         server_tcp_out_port = 4022
5001         server_udp_in_port = 23
5002         server_udp_out_port = 4023
5003         client_tcp_in_port = 1234
5004         client_udp_in_port = 1235
5005         client_tcp_out_port = 0
5006         client_udp_out_port = 0
5007         ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr]))
5008         nat_addr_ip6 = ip.src
5009
5010         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5011                                                 self.nat_addr_n)
5012         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5013         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5014
5015         self.vapi.nat64_add_del_static_bib(server.ip6n,
5016                                            self.nat_addr_n,
5017                                            server_tcp_in_port,
5018                                            server_tcp_out_port,
5019                                            IP_PROTOS.tcp)
5020         self.vapi.nat64_add_del_static_bib(server.ip6n,
5021                                            self.nat_addr_n,
5022                                            server_udp_in_port,
5023                                            server_udp_out_port,
5024                                            IP_PROTOS.udp)
5025
5026         # client to server
5027         pkts = []
5028         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5029              IPv6(src=client.ip6, dst=nat_addr_ip6) /
5030              TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
5031         pkts.append(p)
5032         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5033              IPv6(src=client.ip6, dst=nat_addr_ip6) /
5034              UDP(sport=client_udp_in_port, dport=server_udp_out_port))
5035         pkts.append(p)
5036         self.pg0.add_stream(pkts)
5037         self.pg_enable_capture(self.pg_interfaces)
5038         self.pg_start()
5039         capture = self.pg0.get_capture(len(pkts))
5040         for packet in capture:
5041             try:
5042                 self.assertEqual(packet[IPv6].src, nat_addr_ip6)
5043                 self.assertEqual(packet[IPv6].dst, server.ip6)
5044                 if packet.haslayer(TCP):
5045                     self.assertNotEqual(packet[TCP].sport, client_tcp_in_port)
5046                     self.assertEqual(packet[TCP].dport, server_tcp_in_port)
5047                     self.check_tcp_checksum(packet)
5048                     client_tcp_out_port = packet[TCP].sport
5049                 else:
5050                     self.assertNotEqual(packet[UDP].sport, client_udp_in_port)
5051                     self.assertEqual(packet[UDP].dport, server_udp_in_port)
5052                     self.check_udp_checksum(packet)
5053                     client_udp_out_port = packet[UDP].sport
5054             except:
5055                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
5056                 raise
5057
5058         # server to client
5059         pkts = []
5060         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5061              IPv6(src=server.ip6, dst=nat_addr_ip6) /
5062              TCP(sport=server_tcp_in_port, dport=client_tcp_out_port))
5063         pkts.append(p)
5064         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5065              IPv6(src=server.ip6, dst=nat_addr_ip6) /
5066              UDP(sport=server_udp_in_port, dport=client_udp_out_port))
5067         pkts.append(p)
5068         self.pg0.add_stream(pkts)
5069         self.pg_enable_capture(self.pg_interfaces)
5070         self.pg_start()
5071         capture = self.pg0.get_capture(len(pkts))
5072         for packet in capture:
5073             try:
5074                 self.assertEqual(packet[IPv6].src, nat_addr_ip6)
5075                 self.assertEqual(packet[IPv6].dst, client.ip6)
5076                 if packet.haslayer(TCP):
5077                     self.assertEqual(packet[TCP].sport, server_tcp_out_port)
5078                     self.assertEqual(packet[TCP].dport, client_tcp_in_port)
5079                     self.check_tcp_checksum(packet)
5080                 else:
5081                     self.assertEqual(packet[UDP].sport, server_udp_out_port)
5082                     self.assertEqual(packet[UDP].dport, client_udp_in_port)
5083                     self.check_udp_checksum(packet)
5084             except:
5085                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
5086                 raise
5087
5088         # ICMP error
5089         pkts = []
5090         pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5091                 IPv6(src=client.ip6, dst=nat_addr_ip6) /
5092                 ICMPv6DestUnreach(code=1) /
5093                 packet[IPv6] for packet in capture]
5094         self.pg0.add_stream(pkts)
5095         self.pg_enable_capture(self.pg_interfaces)
5096         self.pg_start()
5097         capture = self.pg0.get_capture(len(pkts))
5098         for packet in capture:
5099             try:
5100                 self.assertEqual(packet[IPv6].src, nat_addr_ip6)
5101                 self.assertEqual(packet[IPv6].dst, server.ip6)
5102                 icmp = packet[ICMPv6DestUnreach]
5103                 self.assertEqual(icmp.code, 1)
5104                 inner = icmp[IPerror6]
5105                 self.assertEqual(inner.src, server.ip6)
5106                 self.assertEqual(inner.dst, nat_addr_ip6)
5107                 self.check_icmpv6_checksum(packet)
5108                 if inner.haslayer(TCPerror):
5109                     self.assertEqual(inner[TCPerror].sport, server_tcp_in_port)
5110                     self.assertEqual(inner[TCPerror].dport,
5111                                      client_tcp_out_port)
5112                 else:
5113                     self.assertEqual(inner[UDPerror].sport, server_udp_in_port)
5114                     self.assertEqual(inner[UDPerror].dport,
5115                                      client_udp_out_port)
5116             except:
5117                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
5118                 raise
5119
5120     def test_prefix(self):
5121         """ NAT64 Network-Specific Prefix """
5122
5123         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5124                                                 self.nat_addr_n)
5125         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5126         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5127         self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
5128                                                 self.vrf1_nat_addr_n,
5129                                                 vrf_id=self.vrf1_id)
5130         self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
5131
5132         # Add global prefix
5133         global_pref64 = "2001:db8::"
5134         global_pref64_n = socket.inet_pton(socket.AF_INET6, global_pref64)
5135         global_pref64_len = 32
5136         self.vapi.nat64_add_del_prefix(global_pref64_n, global_pref64_len)
5137
5138         prefix = self.vapi.nat64_prefix_dump()
5139         self.assertEqual(len(prefix), 1)
5140         self.assertEqual(prefix[0].prefix, global_pref64_n)
5141         self.assertEqual(prefix[0].prefix_len, global_pref64_len)
5142         self.assertEqual(prefix[0].vrf_id, 0)
5143
5144         # Add tenant specific prefix
5145         vrf1_pref64 = "2001:db8:122:300::"
5146         vrf1_pref64_n = socket.inet_pton(socket.AF_INET6, vrf1_pref64)
5147         vrf1_pref64_len = 56
5148         self.vapi.nat64_add_del_prefix(vrf1_pref64_n,
5149                                        vrf1_pref64_len,
5150                                        vrf_id=self.vrf1_id)
5151         prefix = self.vapi.nat64_prefix_dump()
5152         self.assertEqual(len(prefix), 2)
5153
5154         # Global prefix
5155         pkts = self.create_stream_in_ip6(self.pg0,
5156                                          self.pg1,
5157                                          pref=global_pref64,
5158                                          plen=global_pref64_len)
5159         self.pg0.add_stream(pkts)
5160         self.pg_enable_capture(self.pg_interfaces)
5161         self.pg_start()
5162         capture = self.pg1.get_capture(len(pkts))
5163         self.verify_capture_out(capture, nat_ip=self.nat_addr,
5164                                 dst_ip=self.pg1.remote_ip4)
5165
5166         pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
5167         self.pg1.add_stream(pkts)
5168         self.pg_enable_capture(self.pg_interfaces)
5169         self.pg_start()
5170         capture = self.pg0.get_capture(len(pkts))
5171         dst_ip = self.compose_ip6(self.pg1.remote_ip4,
5172                                   global_pref64,
5173                                   global_pref64_len)
5174         self.verify_capture_in_ip6(capture, dst_ip, self.pg0.remote_ip6)
5175
5176         # Tenant specific prefix
5177         pkts = self.create_stream_in_ip6(self.pg2,
5178                                          self.pg1,
5179                                          pref=vrf1_pref64,
5180                                          plen=vrf1_pref64_len)
5181         self.pg2.add_stream(pkts)
5182         self.pg_enable_capture(self.pg_interfaces)
5183         self.pg_start()
5184         capture = self.pg1.get_capture(len(pkts))
5185         self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
5186                                 dst_ip=self.pg1.remote_ip4)
5187
5188         pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
5189         self.pg1.add_stream(pkts)
5190         self.pg_enable_capture(self.pg_interfaces)
5191         self.pg_start()
5192         capture = self.pg2.get_capture(len(pkts))
5193         dst_ip = self.compose_ip6(self.pg1.remote_ip4,
5194                                   vrf1_pref64,
5195                                   vrf1_pref64_len)
5196         self.verify_capture_in_ip6(capture, dst_ip, self.pg2.remote_ip6)
5197
5198     def test_unknown_proto(self):
5199         """ NAT64 translate packet with unknown protocol """
5200
5201         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5202                                                 self.nat_addr_n)
5203         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5204         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5205         remote_ip6 = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
5206
5207         # in2out
5208         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5209              IPv6(src=self.pg0.remote_ip6, dst=remote_ip6) /
5210              TCP(sport=self.tcp_port_in, dport=20))
5211         self.pg0.add_stream(p)
5212         self.pg_enable_capture(self.pg_interfaces)
5213         self.pg_start()
5214         p = self.pg1.get_capture(1)
5215
5216         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5217              IPv6(src=self.pg0.remote_ip6, dst=remote_ip6, nh=47) /
5218              GRE() /
5219              IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
5220              TCP(sport=1234, dport=1234))
5221         self.pg0.add_stream(p)
5222         self.pg_enable_capture(self.pg_interfaces)
5223         self.pg_start()
5224         p = self.pg1.get_capture(1)
5225         packet = p[0]
5226         try:
5227             self.assertEqual(packet[IP].src, self.nat_addr)
5228             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
5229             self.assertTrue(packet.haslayer(GRE))
5230             self.check_ip_checksum(packet)
5231         except:
5232             self.logger.error(ppp("Unexpected or invalid packet:", packet))
5233             raise
5234
5235         # out2in
5236         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
5237              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
5238              GRE() /
5239              IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
5240              TCP(sport=1234, dport=1234))
5241         self.pg1.add_stream(p)
5242         self.pg_enable_capture(self.pg_interfaces)
5243         self.pg_start()
5244         p = self.pg0.get_capture(1)
5245         packet = p[0]
5246         try:
5247             self.assertEqual(packet[IPv6].src, remote_ip6)
5248             self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
5249             self.assertEqual(packet[IPv6].nh, 47)
5250         except:
5251             self.logger.error(ppp("Unexpected or invalid packet:", packet))
5252             raise
5253
5254     def test_hairpinning_unknown_proto(self):
5255         """ NAT64 translate packet with unknown protocol - hairpinning """
5256
5257         client = self.pg0.remote_hosts[0]
5258         server = self.pg0.remote_hosts[1]
5259         server_tcp_in_port = 22
5260         server_tcp_out_port = 4022
5261         client_tcp_in_port = 1234
5262         client_tcp_out_port = 1235
5263         server_nat_ip = "10.0.0.100"
5264         client_nat_ip = "10.0.0.110"
5265         server_nat_ip_n = socket.inet_pton(socket.AF_INET, server_nat_ip)
5266         client_nat_ip_n = socket.inet_pton(socket.AF_INET, client_nat_ip)
5267         server_nat_ip6 = self.compose_ip6(server_nat_ip, '64:ff9b::', 96)
5268         client_nat_ip6 = self.compose_ip6(client_nat_ip, '64:ff9b::', 96)
5269
5270         self.vapi.nat64_add_del_pool_addr_range(server_nat_ip_n,
5271                                                 client_nat_ip_n)
5272         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5273         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5274
5275         self.vapi.nat64_add_del_static_bib(server.ip6n,
5276                                            server_nat_ip_n,
5277                                            server_tcp_in_port,
5278                                            server_tcp_out_port,
5279                                            IP_PROTOS.tcp)
5280
5281         self.vapi.nat64_add_del_static_bib(server.ip6n,
5282                                            server_nat_ip_n,
5283                                            0,
5284                                            0,
5285                                            IP_PROTOS.gre)
5286
5287         self.vapi.nat64_add_del_static_bib(client.ip6n,
5288                                            client_nat_ip_n,
5289                                            client_tcp_in_port,
5290                                            client_tcp_out_port,
5291                                            IP_PROTOS.tcp)
5292
5293         # client to server
5294         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5295              IPv6(src=client.ip6, dst=server_nat_ip6) /
5296              TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
5297         self.pg0.add_stream(p)
5298         self.pg_enable_capture(self.pg_interfaces)
5299         self.pg_start()
5300         p = self.pg0.get_capture(1)
5301
5302         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5303              IPv6(src=client.ip6, dst=server_nat_ip6, nh=IP_PROTOS.gre) /
5304              GRE() /
5305              IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
5306              TCP(sport=1234, dport=1234))
5307         self.pg0.add_stream(p)
5308         self.pg_enable_capture(self.pg_interfaces)
5309         self.pg_start()
5310         p = self.pg0.get_capture(1)
5311         packet = p[0]
5312         try:
5313             self.assertEqual(packet[IPv6].src, client_nat_ip6)
5314             self.assertEqual(packet[IPv6].dst, server.ip6)
5315             self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre)
5316         except:
5317             self.logger.error(ppp("Unexpected or invalid packet:", packet))
5318             raise
5319
5320         # server to client
5321         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5322              IPv6(src=server.ip6, dst=client_nat_ip6, nh=IP_PROTOS.gre) /
5323              GRE() /
5324              IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
5325              TCP(sport=1234, dport=1234))
5326         self.pg0.add_stream(p)
5327         self.pg_enable_capture(self.pg_interfaces)
5328         self.pg_start()
5329         p = self.pg0.get_capture(1)
5330         packet = p[0]
5331         try:
5332             self.assertEqual(packet[IPv6].src, server_nat_ip6)
5333             self.assertEqual(packet[IPv6].dst, client.ip6)
5334             self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre)
5335         except:
5336             self.logger.error(ppp("Unexpected or invalid packet:", packet))
5337             raise
5338
5339     def test_one_armed_nat64(self):
5340         """ One armed NAT64 """
5341         external_port = 0
5342         remote_host_ip6 = self.compose_ip6(self.pg3.remote_ip4,
5343                                            '64:ff9b::',
5344                                            96)
5345
5346         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5347                                                 self.nat_addr_n)
5348         self.vapi.nat64_add_del_interface(self.pg3.sw_if_index)
5349         self.vapi.nat64_add_del_interface(self.pg3.sw_if_index, is_inside=0)
5350
5351         # in2out
5352         p = (Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) /
5353              IPv6(src=self.pg3.remote_ip6, dst=remote_host_ip6) /
5354              TCP(sport=12345, dport=80))
5355         self.pg3.add_stream(p)
5356         self.pg_enable_capture(self.pg_interfaces)
5357         self.pg_start()
5358         capture = self.pg3.get_capture(1)
5359         p = capture[0]
5360         try:
5361             ip = p[IP]
5362             tcp = p[TCP]
5363             self.assertEqual(ip.src, self.nat_addr)
5364             self.assertEqual(ip.dst, self.pg3.remote_ip4)
5365             self.assertNotEqual(tcp.sport, 12345)
5366             external_port = tcp.sport
5367             self.assertEqual(tcp.dport, 80)
5368             self.check_tcp_checksum(p)
5369             self.check_ip_checksum(p)
5370         except:
5371             self.logger.error(ppp("Unexpected or invalid packet:", p))
5372             raise
5373
5374         # out2in
5375         p = (Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) /
5376              IP(src=self.pg3.remote_ip4, dst=self.nat_addr) /
5377              TCP(sport=80, dport=external_port))
5378         self.pg3.add_stream(p)
5379         self.pg_enable_capture(self.pg_interfaces)
5380         self.pg_start()
5381         capture = self.pg3.get_capture(1)
5382         p = capture[0]
5383         try:
5384             ip = p[IPv6]
5385             tcp = p[TCP]
5386             self.assertEqual(ip.src, remote_host_ip6)
5387             self.assertEqual(ip.dst, self.pg3.remote_ip6)
5388             self.assertEqual(tcp.sport, 80)
5389             self.assertEqual(tcp.dport, 12345)
5390             self.check_tcp_checksum(p)
5391         except:
5392             self.logger.error(ppp("Unexpected or invalid packet:", p))
5393             raise
5394
5395     def test_frag_in_order(self):
5396         """ NAT64 translate fragments arriving in order """
5397         self.tcp_port_in = random.randint(1025, 65535)
5398
5399         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5400                                                 self.nat_addr_n)
5401         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5402         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5403
5404         reass = self.vapi.nat_reass_dump()
5405         reass_n_start = len(reass)
5406
5407         # in2out
5408         data = 'a' * 200
5409         pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
5410                                            self.tcp_port_in, 20, data)
5411         self.pg0.add_stream(pkts)
5412         self.pg_enable_capture(self.pg_interfaces)
5413         self.pg_start()
5414         frags = self.pg1.get_capture(len(pkts))
5415         p = self.reass_frags_and_verify(frags,
5416                                         self.nat_addr,
5417                                         self.pg1.remote_ip4)
5418         self.assertEqual(p[TCP].dport, 20)
5419         self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
5420         self.tcp_port_out = p[TCP].sport
5421         self.assertEqual(data, p[Raw].load)
5422
5423         # out2in
5424         data = "A" * 4 + "b" * 16 + "C" * 3
5425         pkts = self.create_stream_frag(self.pg1,
5426                                        self.nat_addr,
5427                                        20,
5428                                        self.tcp_port_out,
5429                                        data)
5430         self.pg1.add_stream(pkts)
5431         self.pg_enable_capture(self.pg_interfaces)
5432         self.pg_start()
5433         frags = self.pg0.get_capture(len(pkts))
5434         src = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
5435         p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6)
5436         self.assertEqual(p[TCP].sport, 20)
5437         self.assertEqual(p[TCP].dport, self.tcp_port_in)
5438         self.assertEqual(data, p[Raw].load)
5439
5440         reass = self.vapi.nat_reass_dump()
5441         reass_n_end = len(reass)
5442
5443         self.assertEqual(reass_n_end - reass_n_start, 2)
5444
5445     def test_reass_hairpinning(self):
5446         """ NAT64 fragments hairpinning """
5447         data = 'a' * 200
5448         client = self.pg0.remote_hosts[0]
5449         server = self.pg0.remote_hosts[1]
5450         server_in_port = random.randint(1025, 65535)
5451         server_out_port = random.randint(1025, 65535)
5452         client_in_port = random.randint(1025, 65535)
5453         ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr]))
5454         nat_addr_ip6 = ip.src
5455
5456         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5457                                                 self.nat_addr_n)
5458         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5459         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5460
5461         # add static BIB entry for server
5462         self.vapi.nat64_add_del_static_bib(server.ip6n,
5463                                            self.nat_addr_n,
5464                                            server_in_port,
5465                                            server_out_port,
5466                                            IP_PROTOS.tcp)
5467
5468         # send packet from host to server
5469         pkts = self.create_stream_frag_ip6(self.pg0,
5470                                            self.nat_addr,
5471                                            client_in_port,
5472                                            server_out_port,
5473                                            data)
5474         self.pg0.add_stream(pkts)
5475         self.pg_enable_capture(self.pg_interfaces)
5476         self.pg_start()
5477         frags = self.pg0.get_capture(len(pkts))
5478         p = self.reass_frags_and_verify_ip6(frags, nat_addr_ip6, server.ip6)
5479         self.assertNotEqual(p[TCP].sport, client_in_port)
5480         self.assertEqual(p[TCP].dport, server_in_port)
5481         self.assertEqual(data, p[Raw].load)
5482
5483     def test_frag_out_of_order(self):
5484         """ NAT64 translate fragments arriving out of order """
5485         self.tcp_port_in = random.randint(1025, 65535)
5486
5487         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5488                                                 self.nat_addr_n)
5489         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5490         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5491
5492         # in2out
5493         data = 'a' * 200
5494         pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
5495                                            self.tcp_port_in, 20, data)
5496         pkts.reverse()
5497         self.pg0.add_stream(pkts)
5498         self.pg_enable_capture(self.pg_interfaces)
5499         self.pg_start()
5500         frags = self.pg1.get_capture(len(pkts))
5501         p = self.reass_frags_and_verify(frags,
5502                                         self.nat_addr,
5503                                         self.pg1.remote_ip4)
5504         self.assertEqual(p[TCP].dport, 20)
5505         self.assertNotEqual(p[TCP].sport, self.tcp_port_in)
5506         self.tcp_port_out = p[TCP].sport
5507         self.assertEqual(data, p[Raw].load)
5508
5509         # out2in
5510         data = "A" * 4 + "B" * 16 + "C" * 3
5511         pkts = self.create_stream_frag(self.pg1,
5512                                        self.nat_addr,
5513                                        20,
5514                                        self.tcp_port_out,
5515                                        data)
5516         pkts.reverse()
5517         self.pg1.add_stream(pkts)
5518         self.pg_enable_capture(self.pg_interfaces)
5519         self.pg_start()
5520         frags = self.pg0.get_capture(len(pkts))
5521         src = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
5522         p = self.reass_frags_and_verify_ip6(frags, src, self.pg0.remote_ip6)
5523         self.assertEqual(p[TCP].sport, 20)
5524         self.assertEqual(p[TCP].dport, self.tcp_port_in)
5525         self.assertEqual(data, p[Raw].load)
5526
5527     def test_interface_addr(self):
5528         """ Acquire NAT64 pool addresses from interface """
5529         self.vapi.nat64_add_interface_addr(self.pg4.sw_if_index)
5530
5531         # no address in NAT64 pool
5532         adresses = self.vapi.nat44_address_dump()
5533         self.assertEqual(0, len(adresses))
5534
5535         # configure interface address and check NAT64 address pool
5536         self.pg4.config_ip4()
5537         addresses = self.vapi.nat64_pool_addr_dump()
5538         self.assertEqual(len(addresses), 1)
5539         self.assertEqual(addresses[0].address, self.pg4.local_ip4n)
5540
5541         # remove interface address and check NAT64 address pool
5542         self.pg4.unconfig_ip4()
5543         addresses = self.vapi.nat64_pool_addr_dump()
5544         self.assertEqual(0, len(adresses))
5545
5546     @unittest.skipUnless(running_extended_tests(), "part of extended tests")
5547     def test_ipfix_max_bibs_sessions(self):
5548         """ IPFIX logging maximum session and BIB entries exceeded """
5549         max_bibs = 1280
5550         max_sessions = 2560
5551         remote_host_ip6 = self.compose_ip6(self.pg1.remote_ip4,
5552                                            '64:ff9b::',
5553                                            96)
5554
5555         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5556                                                 self.nat_addr_n)
5557         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5558         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5559
5560         pkts = []
5561         src = ""
5562         for i in range(0, max_bibs):
5563             src = "fd01:aa::%x" % (i)
5564             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
5565                  IPv6(src=src, dst=remote_host_ip6) /
5566                  TCP(sport=12345, dport=80))
5567             pkts.append(p)
5568             p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
5569                  IPv6(src=src, dst=remote_host_ip6) /
5570                  TCP(sport=12345, dport=22))
5571             pkts.append(p)
5572         self.pg0.add_stream(pkts)
5573         self.pg_enable_capture(self.pg_interfaces)
5574         self.pg_start()
5575         self.pg1.get_capture(max_sessions)
5576
5577         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
5578                                      src_address=self.pg3.local_ip4n,
5579                                      path_mtu=512,
5580                                      template_interval=10)
5581         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
5582                             src_port=self.ipfix_src_port)
5583
5584         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
5585              IPv6(src=src, dst=remote_host_ip6) /
5586              TCP(sport=12345, dport=25))
5587         self.pg0.add_stream(p)
5588         self.pg_enable_capture(self.pg_interfaces)
5589         self.pg_start()
5590         self.pg1.get_capture(0)
5591         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
5592         capture = self.pg3.get_capture(9)
5593         ipfix = IPFIXDecoder()
5594         # first load template
5595         for p in capture:
5596             self.assertTrue(p.haslayer(IPFIX))
5597             self.assertEqual(p[IP].src, self.pg3.local_ip4)
5598             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
5599             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
5600             self.assertEqual(p[UDP].dport, 4739)
5601             self.assertEqual(p[IPFIX].observationDomainID,
5602                              self.ipfix_domain_id)
5603             if p.haslayer(Template):
5604                 ipfix.add_template(p.getlayer(Template))
5605         # verify events in data set
5606         for p in capture:
5607             if p.haslayer(Data):
5608                 data = ipfix.decode_data_set(p.getlayer(Set))
5609                 self.verify_ipfix_max_sessions(data, max_sessions)
5610
5611         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
5612              IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6) /
5613              TCP(sport=12345, dport=80))
5614         self.pg0.add_stream(p)
5615         self.pg_enable_capture(self.pg_interfaces)
5616         self.pg_start()
5617         self.pg1.get_capture(0)
5618         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
5619         capture = self.pg3.get_capture(1)
5620         # verify events in data set
5621         for p in capture:
5622             self.assertTrue(p.haslayer(IPFIX))
5623             self.assertEqual(p[IP].src, self.pg3.local_ip4)
5624             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
5625             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
5626             self.assertEqual(p[UDP].dport, 4739)
5627             self.assertEqual(p[IPFIX].observationDomainID,
5628                              self.ipfix_domain_id)
5629             if p.haslayer(Data):
5630                 data = ipfix.decode_data_set(p.getlayer(Set))
5631                 self.verify_ipfix_max_bibs(data, max_bibs)
5632
5633     def test_ipfix_max_frags(self):
5634         """ IPFIX logging maximum fragments pending reassembly exceeded """
5635         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5636                                                 self.nat_addr_n)
5637         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5638         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5639         self.vapi.nat_set_reass(max_frag=0, is_ip6=1)
5640         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
5641                                      src_address=self.pg3.local_ip4n,
5642                                      path_mtu=512,
5643                                      template_interval=10)
5644         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
5645                             src_port=self.ipfix_src_port)
5646
5647         data = 'a' * 200
5648         pkts = self.create_stream_frag_ip6(self.pg0, self.pg1.remote_ip4,
5649                                            self.tcp_port_in, 20, data)
5650         self.pg0.add_stream(pkts[-1])
5651         self.pg_enable_capture(self.pg_interfaces)
5652         self.pg_start()
5653         self.pg1.get_capture(0)
5654         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
5655         capture = self.pg3.get_capture(9)
5656         ipfix = IPFIXDecoder()
5657         # first load template
5658         for p in capture:
5659             self.assertTrue(p.haslayer(IPFIX))
5660             self.assertEqual(p[IP].src, self.pg3.local_ip4)
5661             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
5662             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
5663             self.assertEqual(p[UDP].dport, 4739)
5664             self.assertEqual(p[IPFIX].observationDomainID,
5665                              self.ipfix_domain_id)
5666             if p.haslayer(Template):
5667                 ipfix.add_template(p.getlayer(Template))
5668         # verify events in data set
5669         for p in capture:
5670             if p.haslayer(Data):
5671                 data = ipfix.decode_data_set(p.getlayer(Set))
5672                 self.verify_ipfix_max_fragments_ip6(data, 0,
5673                                                     self.pg0.remote_ip6n)
5674
5675     def test_ipfix_bib_ses(self):
5676         """ IPFIX logging NAT64 BIB/session create and delete events """
5677         self.tcp_port_in = random.randint(1025, 65535)
5678         remote_host_ip6 = self.compose_ip6(self.pg1.remote_ip4,
5679                                            '64:ff9b::',
5680                                            96)
5681
5682         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5683                                                 self.nat_addr_n)
5684         self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
5685         self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
5686         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
5687                                      src_address=self.pg3.local_ip4n,
5688                                      path_mtu=512,
5689                                      template_interval=10)
5690         self.vapi.nat_ipfix(domain_id=self.ipfix_domain_id,
5691                             src_port=self.ipfix_src_port)
5692
5693         # Create
5694         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
5695              IPv6(src=self.pg0.remote_ip6, dst=remote_host_ip6) /
5696              TCP(sport=self.tcp_port_in, dport=25))
5697         self.pg0.add_stream(p)
5698         self.pg_enable_capture(self.pg_interfaces)
5699         self.pg_start()
5700         p = self.pg1.get_capture(1)
5701         self.tcp_port_out = p[0][TCP].sport
5702         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
5703         capture = self.pg3.get_capture(10)
5704         ipfix = IPFIXDecoder()
5705         # first load template
5706         for p in capture:
5707             self.assertTrue(p.haslayer(IPFIX))
5708             self.assertEqual(p[IP].src, self.pg3.local_ip4)
5709             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
5710             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
5711             self.assertEqual(p[UDP].dport, 4739)
5712             self.assertEqual(p[IPFIX].observationDomainID,
5713                              self.ipfix_domain_id)
5714             if p.haslayer(Template):
5715                 ipfix.add_template(p.getlayer(Template))
5716         # verify events in data set
5717         for p in capture:
5718             if p.haslayer(Data):
5719                 data = ipfix.decode_data_set(p.getlayer(Set))
5720                 if ord(data[0][230]) == 10:
5721                     self.verify_ipfix_bib(data, 1, self.pg0.remote_ip6n)
5722                 elif ord(data[0][230]) == 6:
5723                     self.verify_ipfix_nat64_ses(data,
5724                                                 1,
5725                                                 self.pg0.remote_ip6n,
5726                                                 self.pg1.remote_ip4,
5727                                                 25)
5728                 else:
5729                     self.logger.error(ppp("Unexpected or invalid packet: ", p))
5730
5731         # Delete
5732         self.pg_enable_capture(self.pg_interfaces)
5733         self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
5734                                                 self.nat_addr_n,
5735                                                 is_add=0)
5736         self.vapi.cli("ipfix flush")  # FIXME this should be an API call
5737         capture = self.pg3.get_capture(2)
5738         # verify events in data set
5739         for p in capture:
5740             self.assertTrue(p.haslayer(IPFIX))
5741             self.assertEqual(p[IP].src, self.pg3.local_ip4)
5742             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
5743             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
5744             self.assertEqual(p[UDP].dport, 4739)
5745             self.assertEqual(p[IPFIX].observationDomainID,
5746                              self.ipfix_domain_id)
5747             if p.haslayer(Data):
5748                 data = ipfix.decode_data_set(p.getlayer(Set))
5749                 if ord(data[0][230]) == 11:
5750                     self.verify_ipfix_bib(data, 0, self.pg0.remote_ip6n)
5751                 elif ord(data[0][230]) == 7:
5752                     self.verify_ipfix_nat64_ses(data,
5753                                                 0,
5754                                                 self.pg0.remote_ip6n,
5755                                                 self.pg1.remote_ip4,
5756                                                 25)
5757                 else:
5758                     self.logger.error(ppp("Unexpected or invalid packet: ", p))
5759
5760     def nat64_get_ses_num(self):
5761         """
5762         Return number of active NAT64 sessions.
5763         """
5764         st = self.vapi.nat64_st_dump()
5765         return len(st)
5766
5767     def clear_nat64(self):
5768         """
5769         Clear NAT64 configuration.
5770         """
5771         self.vapi.nat_ipfix(enable=0, src_port=self.ipfix_src_port,
5772                             domain_id=self.ipfix_domain_id)
5773         self.ipfix_src_port = 4739
5774         self.ipfix_domain_id = 1
5775
5776         self.vapi.nat64_set_timeouts()
5777
5778         interfaces = self.vapi.nat64_interface_dump()
5779         for intf in interfaces:
5780             if intf.is_inside > 1:
5781                 self.vapi.nat64_add_del_interface(intf.sw_if_index,
5782                                                   0,
5783                                                   is_add=0)
5784             self.vapi.nat64_add_del_interface(intf.sw_if_index,
5785                                               intf.is_inside,
5786                                               is_add=0)
5787
5788         bib = self.vapi.nat64_bib_dump(255)
5789         for bibe in bib:
5790             if bibe.is_static:
5791                 self.vapi.nat64_add_del_static_bib(bibe.i_addr,
5792                                                    bibe.o_addr,
5793                                                    bibe.i_port,
5794                                                    bibe.o_port,
5795                                                    bibe.proto,
5796                                                    bibe.vrf_id,
5797                                                    is_add=0)
5798
5799         adresses = self.vapi.nat64_pool_addr_dump()
5800         for addr in adresses:
5801             self.vapi.nat64_add_del_pool_addr_range(addr.address,
5802                                                     addr.address,
5803                                                     vrf_id=addr.vrf_id,
5804                                                     is_add=0)
5805
5806         prefixes = self.vapi.nat64_prefix_dump()
5807         for prefix in prefixes:
5808             self.vapi.nat64_add_del_prefix(prefix.prefix,
5809                                            prefix.prefix_len,
5810                                            vrf_id=prefix.vrf_id,
5811                                            is_add=0)
5812
5813     def tearDown(self):
5814         super(TestNAT64, self).tearDown()
5815         if not self.vpp_dead:
5816             self.logger.info(self.vapi.cli("show nat64 pool"))
5817             self.logger.info(self.vapi.cli("show nat64 interfaces"))
5818             self.logger.info(self.vapi.cli("show nat64 prefix"))
5819             self.logger.info(self.vapi.cli("show nat64 bib all"))
5820             self.logger.info(self.vapi.cli("show nat64 session table all"))
5821             self.logger.info(self.vapi.cli("show nat virtual-reassembly"))
5822             self.clear_nat64()
5823
5824
5825 class TestDSlite(MethodHolder):
5826     """ DS-Lite Test Cases """
5827
5828     @classmethod
5829     def setUpClass(cls):
5830         super(TestDSlite, cls).setUpClass()
5831
5832         try:
5833             cls.nat_addr = '10.0.0.3'
5834             cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
5835
5836             cls.create_pg_interfaces(range(2))
5837             cls.pg0.admin_up()
5838             cls.pg0.config_ip4()
5839             cls.pg0.resolve_arp()
5840             cls.pg1.admin_up()
5841             cls.pg1.config_ip6()
5842             cls.pg1.generate_remote_hosts(2)
5843             cls.pg1.configure_ipv6_neighbors()
5844
5845         except Exception:
5846             super(TestDSlite, cls).tearDownClass()
5847             raise
5848
5849     def test_dslite(self):
5850         """ Test DS-Lite """
5851         self.vapi.dslite_add_del_pool_addr_range(self.nat_addr_n,
5852                                                  self.nat_addr_n)
5853         aftr_ip4 = '192.0.0.1'
5854         aftr_ip4_n = socket.inet_pton(socket.AF_INET, aftr_ip4)
5855         aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
5856         aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
5857         self.vapi.dslite_set_aftr_addr(aftr_ip6_n, aftr_ip4_n)
5858
5859         # UDP
5860         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
5861              IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
5862              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
5863              UDP(sport=20000, dport=10000))
5864         self.pg1.add_stream(p)
5865         self.pg_enable_capture(self.pg_interfaces)
5866         self.pg_start()
5867         capture = self.pg0.get_capture(1)
5868         capture = capture[0]
5869         self.assertFalse(capture.haslayer(IPv6))
5870         self.assertEqual(capture[IP].src, self.nat_addr)
5871         self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
5872         self.assertNotEqual(capture[UDP].sport, 20000)
5873         self.assertEqual(capture[UDP].dport, 10000)
5874         self.check_ip_checksum(capture)
5875         out_port = capture[UDP].sport
5876
5877         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5878              IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
5879              UDP(sport=10000, dport=out_port))
5880         self.pg0.add_stream(p)
5881         self.pg_enable_capture(self.pg_interfaces)
5882         self.pg_start()
5883         capture = self.pg1.get_capture(1)
5884         capture = capture[0]
5885         self.assertEqual(capture[IPv6].src, aftr_ip6)
5886         self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
5887         self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
5888         self.assertEqual(capture[IP].dst, '192.168.1.1')
5889         self.assertEqual(capture[UDP].sport, 10000)
5890         self.assertEqual(capture[UDP].dport, 20000)
5891         self.check_ip_checksum(capture)
5892
5893         # TCP
5894         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
5895              IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
5896              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
5897              TCP(sport=20001, dport=10001))
5898         self.pg1.add_stream(p)
5899         self.pg_enable_capture(self.pg_interfaces)
5900         self.pg_start()
5901         capture = self.pg0.get_capture(1)
5902         capture = capture[0]
5903         self.assertFalse(capture.haslayer(IPv6))
5904         self.assertEqual(capture[IP].src, self.nat_addr)
5905         self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
5906         self.assertNotEqual(capture[TCP].sport, 20001)
5907         self.assertEqual(capture[TCP].dport, 10001)
5908         self.check_ip_checksum(capture)
5909         self.check_tcp_checksum(capture)
5910         out_port = capture[TCP].sport
5911
5912         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5913              IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
5914              TCP(sport=10001, dport=out_port))
5915         self.pg0.add_stream(p)
5916         self.pg_enable_capture(self.pg_interfaces)
5917         self.pg_start()
5918         capture = self.pg1.get_capture(1)
5919         capture = capture[0]
5920         self.assertEqual(capture[IPv6].src, aftr_ip6)
5921         self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
5922         self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
5923         self.assertEqual(capture[IP].dst, '192.168.1.1')
5924         self.assertEqual(capture[TCP].sport, 10001)
5925         self.assertEqual(capture[TCP].dport, 20001)
5926         self.check_ip_checksum(capture)
5927         self.check_tcp_checksum(capture)
5928
5929         # ICMP
5930         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
5931              IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
5932              IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
5933              ICMP(id=4000, type='echo-request'))
5934         self.pg1.add_stream(p)
5935         self.pg_enable_capture(self.pg_interfaces)
5936         self.pg_start()
5937         capture = self.pg0.get_capture(1)
5938         capture = capture[0]
5939         self.assertFalse(capture.haslayer(IPv6))
5940         self.assertEqual(capture[IP].src, self.nat_addr)
5941         self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
5942         self.assertNotEqual(capture[ICMP].id, 4000)
5943         self.check_ip_checksum(capture)
5944         self.check_icmp_checksum(capture)
5945         out_id = capture[ICMP].id
5946
5947         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
5948              IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
5949              ICMP(id=out_id, type='echo-reply'))
5950         self.pg0.add_stream(p)
5951         self.pg_enable_capture(self.pg_interfaces)
5952         self.pg_start()
5953         capture = self.pg1.get_capture(1)
5954         capture = capture[0]
5955         self.assertEqual(capture[IPv6].src, aftr_ip6)
5956         self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
5957         self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
5958         self.assertEqual(capture[IP].dst, '192.168.1.1')
5959         self.assertEqual(capture[ICMP].id, 4000)
5960         self.check_ip_checksum(capture)
5961         self.check_icmp_checksum(capture)
5962
5963         # ping DS-Lite AFTR tunnel endpoint address
5964         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
5965              IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6) /
5966              ICMPv6EchoRequest())
5967         self.pg1.add_stream(p)
5968         self.pg_enable_capture(self.pg_interfaces)
5969         self.pg_start()
5970         capture = self.pg1.get_capture(1)
5971         self.assertEqual(1, len(capture))
5972         capture = capture[0]
5973         self.assertEqual(capture[IPv6].src, aftr_ip6)
5974         self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
5975         self.assertTrue(capture.haslayer(ICMPv6EchoReply))
5976
5977     def tearDown(self):
5978         super(TestDSlite, self).tearDown()
5979         if not self.vpp_dead:
5980             self.logger.info(self.vapi.cli("show dslite pool"))
5981             self.logger.info(
5982                 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
5983             self.logger.info(self.vapi.cli("show dslite sessions"))
5984
5985
5986 class TestDSliteCE(MethodHolder):
5987     """ DS-Lite CE Test Cases """
5988
5989     @classmethod
5990     def setUpConstants(cls):
5991         super(TestDSliteCE, cls).setUpConstants()
5992         cls.vpp_cmdline.extend(["nat", "{", "dslite ce", "}"])
5993
5994     @classmethod
5995     def setUpClass(cls):
5996         super(TestDSliteCE, cls).setUpClass()
5997
5998         try:
5999             cls.create_pg_interfaces(range(2))
6000             cls.pg0.admin_up()
6001             cls.pg0.config_ip4()
6002             cls.pg0.resolve_arp()
6003             cls.pg1.admin_up()
6004             cls.pg1.config_ip6()
6005             cls.pg1.generate_remote_hosts(1)
6006             cls.pg1.configure_ipv6_neighbors()
6007
6008         except Exception:
6009             super(TestDSliteCE, cls).tearDownClass()
6010             raise
6011
6012     def test_dslite_ce(self):
6013         """ Test DS-Lite CE """
6014
6015         b4_ip4 = '192.0.0.2'
6016         b4_ip4_n = socket.inet_pton(socket.AF_INET, b4_ip4)
6017         b4_ip6 = '2001:db8:62aa::375e:f4c1:1'
6018         b4_ip6_n = socket.inet_pton(socket.AF_INET6, b4_ip6)
6019         self.vapi.dslite_set_b4_addr(b4_ip6_n, b4_ip4_n)
6020
6021         aftr_ip4 = '192.0.0.1'
6022         aftr_ip4_n = socket.inet_pton(socket.AF_INET, aftr_ip4)
6023         aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
6024         aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
6025         self.vapi.dslite_set_aftr_addr(aftr_ip6_n, aftr_ip4_n)
6026
6027         self.vapi.ip_add_del_route(dst_address=aftr_ip6_n,
6028                                    dst_address_length=128,
6029                                    next_hop_address=self.pg1.remote_ip6n,
6030                                    next_hop_sw_if_index=self.pg1.sw_if_index,
6031                                    is_ipv6=1)
6032
6033         # UDP encapsulation
6034         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
6035              IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4) /
6036              UDP(sport=10000, dport=20000))
6037         self.pg0.add_stream(p)
6038         self.pg_enable_capture(self.pg_interfaces)
6039         self.pg_start()
6040         capture = self.pg1.get_capture(1)
6041         capture = capture[0]
6042         self.assertEqual(capture[IPv6].src, b4_ip6)
6043         self.assertEqual(capture[IPv6].dst, aftr_ip6)
6044         self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
6045         self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
6046         self.assertEqual(capture[UDP].sport, 10000)
6047         self.assertEqual(capture[UDP].dport, 20000)
6048         self.check_ip_checksum(capture)
6049
6050         # UDP decapsulation
6051         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
6052              IPv6(dst=b4_ip6, src=aftr_ip6) /
6053              IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4) /
6054              UDP(sport=20000, dport=10000))
6055         self.pg1.add_stream(p)
6056         self.pg_enable_capture(self.pg_interfaces)
6057         self.pg_start()
6058         capture = self.pg0.get_capture(1)
6059         capture = capture[0]
6060         self.assertFalse(capture.haslayer(IPv6))
6061         self.assertEqual(capture[IP].src, self.pg1.remote_ip4)
6062         self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
6063         self.assertEqual(capture[UDP].sport, 20000)
6064         self.assertEqual(capture[UDP].dport, 10000)
6065         self.check_ip_checksum(capture)
6066
6067         # ping DS-Lite B4 tunnel endpoint address
6068         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
6069              IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6) /
6070              ICMPv6EchoRequest())
6071         self.pg1.add_stream(p)
6072         self.pg_enable_capture(self.pg_interfaces)
6073         self.pg_start()
6074         capture = self.pg1.get_capture(1)
6075         self.assertEqual(1, len(capture))
6076         capture = capture[0]
6077         self.assertEqual(capture[IPv6].src, b4_ip6)
6078         self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
6079         self.assertTrue(capture.haslayer(ICMPv6EchoReply))
6080
6081     def tearDown(self):
6082         super(TestDSliteCE, self).tearDown()
6083         if not self.vpp_dead:
6084             self.logger.info(
6085                 self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
6086             self.logger.info(
6087                 self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
6088
6089 if __name__ == '__main__':
6090     unittest.main(testRunner=VppTestRunner)