src/plugins/nat/test/test_nat44_ei.py

   1 #!/usr/bin/env python3
   2
   3 import ipaddress
   4 import random
   5 import socket
   6 import struct
   7 import unittest
   8 from io import BytesIO
   9 from time import sleep
  10
  11 import scapy.compat
  12 from framework import tag_fixme_vpp_workers
  13 from framework import VppTestCase, VppTestRunner
  14 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
  15 from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
  16     IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
  17     PacketListField
  18 from scapy.data import IP_PROTOS
  19 from scapy.layers.inet import IP, TCP, UDP, ICMP
  20 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
  21 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
  22 from scapy.layers.l2 import Ether, ARP, GRE
  23 from scapy.packet import Raw
  24 from syslog_rfc5424_parser import SyslogMessage, ParseError
  25 from syslog_rfc5424_parser.constants import SyslogSeverity
  26 from util import ppp
  27 from vpp_ip_route import VppIpRoute, VppRoutePath
  28 from vpp_neighbor import VppNeighbor
  29 from vpp_papi import VppEnum
  30
  31
  32 # NAT HA protocol event data
  33 class Event(Packet):
  34     name = "Event"
  35     fields_desc = [ByteEnumField("event_type", None,
  36                                  {1: "add", 2: "del", 3: "refresh"}),
  37                    ByteEnumField("protocol", None,
  38                                  {0: "other", 1: "udp", 2: "tcp", 3: "icmp"}),
  39                    ShortField("flags", 0),
  40                    IPField("in_addr", None),
  41                    IPField("out_addr", None),
  42                    ShortField("in_port", None),
  43                    ShortField("out_port", None),
  44                    IPField("eh_addr", None),
  45                    IPField("ehn_addr", None),
  46                    ShortField("eh_port", None),
  47                    ShortField("ehn_port", None),
  48                    IntField("fib_index", None),
  49                    IntField("total_pkts", 0),
  50                    LongField("total_bytes", 0)]
  51
  52     def extract_padding(self, s):
  53         return "", s
  54
  55
  56 # NAT HA protocol header
  57 class HANATStateSync(Packet):
  58     name = "HA NAT state sync"
  59     fields_desc = [XByteField("version", 1),
  60                    FlagsField("flags", 0, 8, ['ACK']),
  61                    FieldLenField("count", None, count_of="events"),
  62                    IntField("sequence_number", 1),
  63                    IntField("thread_index", 0),
  64                    PacketListField("events", [], Event,
  65                                    count_from=lambda pkt: pkt.count)]
  66
  67
  68 class MethodHolder(VppTestCase):
  69     """ NAT create capture and verify method holder """
  70
  71     @property
  72     def config_flags(self):
  73         return VppEnum.vl_api_nat44_ei_config_flags_t
  74
  75     @property
  76     def SYSLOG_SEVERITY(self):
  77         return VppEnum.vl_api_syslog_severity_t
  78
  79     def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
  80                                  local_port=0, external_port=0, vrf_id=0,
  81                                  is_add=1, external_sw_if_index=0xFFFFFFFF,
  82                                  proto=0, tag="", flags=0):
  83         """
  84         Add/delete NAT44EI static mapping
  85
  86         :param local_ip: Local IP address
  87         :param external_ip: External IP address
  88         :param local_port: Local port number (Optional)
  89         :param external_port: External port number (Optional)
  90         :param vrf_id: VRF ID (Default 0)
  91         :param is_add: 1 if add, 0 if delete (Default add)
  92         :param external_sw_if_index: External interface instead of IP address
  93         :param proto: IP protocol (Mandatory if port specified)
  94         :param tag: Opaque string tag
  95         :param flags: NAT configuration flags
  96         """
  97
  98         if not (local_port and external_port):
  99             flags |= self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
 100
 101         self.vapi.nat44_ei_add_del_static_mapping(
 102             is_add=is_add,
 103             local_ip_address=local_ip,
 104             external_ip_address=external_ip,
 105             external_sw_if_index=external_sw_if_index,
 106             local_port=local_port,
 107             external_port=external_port,
 108             vrf_id=vrf_id, protocol=proto,
 109             flags=flags,
 110             tag=tag)
 111
 112     def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
 113         """
 114         Add/delete NAT44EI address
 115
 116         :param ip: IP address
 117         :param is_add: 1 if add, 0 if delete (Default add)
 118         """
 119         self.vapi.nat44_ei_add_del_address_range(first_ip_address=ip,
 120                                                  last_ip_address=ip,
 121                                                  vrf_id=vrf_id,
 122                                                  is_add=is_add)
 123
 124     def create_routes_and_neigbors(self):
 125         r1 = VppIpRoute(self, self.pg7.remote_ip4, 32,
 126                         [VppRoutePath(self.pg7.remote_ip4,
 127                                       self.pg7.sw_if_index)])
 128         r2 = VppIpRoute(self, self.pg8.remote_ip4, 32,
 129                         [VppRoutePath(self.pg8.remote_ip4,
 130                                       self.pg8.sw_if_index)])
 131         r1.add_vpp_config()
 132         r2.add_vpp_config()
 133
 134         n1 = VppNeighbor(self,
 135                          self.pg7.sw_if_index,
 136                          self.pg7.remote_mac,
 137                          self.pg7.remote_ip4,
 138                          is_static=1)
 139         n2 = VppNeighbor(self,
 140                          self.pg8.sw_if_index,
 141                          self.pg8.remote_mac,
 142                          self.pg8.remote_ip4,
 143                          is_static=1)
 144         n1.add_vpp_config()
 145         n2.add_vpp_config()
 146
 147     def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
 148         """
 149         Create packet stream for inside network
 150
 151         :param in_if: Inside interface
 152         :param out_if: Outside interface
 153         :param dst_ip: Destination address
 154         :param ttl: TTL of generated packets
 155         """
 156         if dst_ip is None:
 157             dst_ip = out_if.remote_ip4
 158
 159         pkts = []
 160         # TCP
 161         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 162              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 163              TCP(sport=self.tcp_port_in, dport=20))
 164         pkts.extend([p, p])
 165
 166         # UDP
 167         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 168              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 169              UDP(sport=self.udp_port_in, dport=20))
 170         pkts.append(p)
 171
 172         # ICMP
 173         p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
 174              IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 175              ICMP(id=self.icmp_id_in, type='echo-request'))
 176         pkts.append(p)
 177
 178         return pkts
 179
 180     def compose_ip6(self, ip4, pref, plen):
 181         """
 182         Compose IPv4-embedded IPv6 addresses
 183
 184         :param ip4: IPv4 address
 185         :param pref: IPv6 prefix
 186         :param plen: IPv6 prefix length
 187         :returns: IPv4-embedded IPv6 addresses
 188         """
 189         pref_n = list(socket.inet_pton(socket.AF_INET6, pref))
 190         ip4_n = list(socket.inet_pton(socket.AF_INET, ip4))
 191         if plen == 32:
 192             pref_n[4] = ip4_n[0]
 193             pref_n[5] = ip4_n[1]
 194             pref_n[6] = ip4_n[2]
 195             pref_n[7] = ip4_n[3]
 196         elif plen == 40:
 197             pref_n[5] = ip4_n[0]
 198             pref_n[6] = ip4_n[1]
 199             pref_n[7] = ip4_n[2]
 200             pref_n[9] = ip4_n[3]
 201         elif plen == 48:
 202             pref_n[6] = ip4_n[0]
 203             pref_n[7] = ip4_n[1]
 204             pref_n[9] = ip4_n[2]
 205             pref_n[10] = ip4_n[3]
 206         elif plen == 56:
 207             pref_n[7] = ip4_n[0]
 208             pref_n[9] = ip4_n[1]
 209             pref_n[10] = ip4_n[2]
 210             pref_n[11] = ip4_n[3]
 211         elif plen == 64:
 212             pref_n[9] = ip4_n[0]
 213             pref_n[10] = ip4_n[1]
 214             pref_n[11] = ip4_n[2]
 215             pref_n[12] = ip4_n[3]
 216         elif plen == 96:
 217             pref_n[12] = ip4_n[0]
 218             pref_n[13] = ip4_n[1]
 219             pref_n[14] = ip4_n[2]
 220             pref_n[15] = ip4_n[3]
 221         packed_pref_n = b''.join([scapy.compat.chb(x) for x in pref_n])
 222         return socket.inet_ntop(socket.AF_INET6, packed_pref_n)
 223
 224     def create_stream_out(self, out_if, dst_ip=None, ttl=64,
 225                           use_inside_ports=False):
 226         """
 227         Create packet stream for outside network
 228
 229         :param out_if: Outside interface
 230         :param dst_ip: Destination IP address (Default use global NAT address)
 231         :param ttl: TTL of generated packets
 232         :param use_inside_ports: Use inside NAT ports as destination ports
 233                instead of outside ports
 234         """
 235         if dst_ip is None:
 236             dst_ip = self.nat_addr
 237         if not use_inside_ports:
 238             tcp_port = self.tcp_port_out
 239             udp_port = self.udp_port_out
 240             icmp_id = self.icmp_id_out
 241         else:
 242             tcp_port = self.tcp_port_in
 243             udp_port = self.udp_port_in
 244             icmp_id = self.icmp_id_in
 245         pkts = []
 246         # TCP
 247         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 248              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 249              TCP(dport=tcp_port, sport=20))
 250         pkts.extend([p, p])
 251
 252         # UDP
 253         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 254              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 255              UDP(dport=udp_port, sport=20))
 256         pkts.append(p)
 257
 258         # ICMP
 259         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 260              IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
 261              ICMP(id=icmp_id, type='echo-reply'))
 262         pkts.append(p)
 263
 264         return pkts
 265
 266     def create_stream_out_ip6(self, out_if, src_ip, dst_ip, hl=64):
 267         """
 268         Create packet stream for outside network
 269
 270         :param out_if: Outside interface
 271         :param dst_ip: Destination IP address (Default use global NAT address)
 272         :param hl: HL of generated packets
 273         """
 274         pkts = []
 275         # TCP
 276         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 277              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 278              TCP(dport=self.tcp_port_out, sport=20))
 279         pkts.append(p)
 280
 281         # UDP
 282         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 283              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 284              UDP(dport=self.udp_port_out, sport=20))
 285         pkts.append(p)
 286
 287         # ICMP
 288         p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
 289              IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
 290              ICMPv6EchoReply(id=self.icmp_id_out))
 291         pkts.append(p)
 292
 293         return pkts
 294
 295     def verify_capture_out(self, capture, nat_ip=None, same_port=False,
 296                            dst_ip=None, is_ip6=False, ignore_port=False):
 297         """
 298         Verify captured packets on outside network
 299
 300         :param capture: Captured packets
 301         :param nat_ip: Translated IP address (Default use global NAT address)
 302         :param same_port: Source port number is not translated (Default False)
 303         :param dst_ip: Destination IP address (Default do not verify)
 304         :param is_ip6: If L3 protocol is IPv6 (Default False)
 305         """
 306         if is_ip6:
 307             IP46 = IPv6
 308             ICMP46 = ICMPv6EchoRequest
 309         else:
 310             IP46 = IP
 311             ICMP46 = ICMP
 312         if nat_ip is None:
 313             nat_ip = self.nat_addr
 314         for packet in capture:
 315             try:
 316                 if not is_ip6:
 317                     self.assert_packet_checksums_valid(packet)
 318                 self.assertEqual(packet[IP46].src, nat_ip)
 319                 if dst_ip is not None:
 320                     self.assertEqual(packet[IP46].dst, dst_ip)
 321                 if packet.haslayer(TCP):
 322                     if not ignore_port:
 323                         if same_port:
 324                             self.assertEqual(
 325                                 packet[TCP].sport, self.tcp_port_in)
 326                         else:
 327                             self.assertNotEqual(
 328                                 packet[TCP].sport, self.tcp_port_in)
 329                     self.tcp_port_out = packet[TCP].sport
 330                     self.assert_packet_checksums_valid(packet)
 331                 elif packet.haslayer(UDP):
 332                     if not ignore_port:
 333                         if same_port:
 334                             self.assertEqual(
 335                                 packet[UDP].sport, self.udp_port_in)
 336                         else:
 337                             self.assertNotEqual(
 338                                 packet[UDP].sport, self.udp_port_in)
 339                     self.udp_port_out = packet[UDP].sport
 340                 else:
 341                     if not ignore_port:
 342                         if same_port:
 343                             self.assertEqual(
 344                                 packet[ICMP46].id, self.icmp_id_in)
 345                         else:
 346                             self.assertNotEqual(
 347                                 packet[ICMP46].id, self.icmp_id_in)
 348                     self.icmp_id_out = packet[ICMP46].id
 349                     self.assert_packet_checksums_valid(packet)
 350             except:
 351                 self.logger.error(ppp("Unexpected or invalid packet "
 352                                       "(outside network):", packet))
 353                 raise
 354
 355     def verify_capture_out_ip6(self, capture, nat_ip, same_port=False,
 356                                dst_ip=None):
 357         """
 358         Verify captured packets on outside network
 359
 360         :param capture: Captured packets
 361         :param nat_ip: Translated IP address
 362         :param same_port: Source port number is not translated (Default False)
 363         :param dst_ip: Destination IP address (Default do not verify)
 364         """
 365         return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
 366                                        True)
 367
 368     def verify_capture_in(self, capture, in_if):
 369         """
 370         Verify captured packets on inside network
 371
 372         :param capture: Captured packets
 373         :param in_if: Inside interface
 374         """
 375         for packet in capture:
 376             try:
 377                 self.assert_packet_checksums_valid(packet)
 378                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 379                 if packet.haslayer(TCP):
 380                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
 381                 elif packet.haslayer(UDP):
 382                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
 383                 else:
 384                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 385             except:
 386                 self.logger.error(ppp("Unexpected or invalid packet "
 387                                       "(inside network):", packet))
 388                 raise
 389
 390     def verify_capture_no_translation(self, capture, ingress_if, egress_if):
 391         """
 392         Verify captured packet that don't have to be translated
 393
 394         :param capture: Captured packets
 395         :param ingress_if: Ingress interface
 396         :param egress_if: Egress interface
 397         """
 398         for packet in capture:
 399             try:
 400                 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
 401                 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
 402                 if packet.haslayer(TCP):
 403                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
 404                 elif packet.haslayer(UDP):
 405                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
 406                 else:
 407                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
 408             except:
 409                 self.logger.error(ppp("Unexpected or invalid packet "
 410                                       "(inside network):", packet))
 411                 raise
 412
 413     def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
 414                                             icmp_type=11):
 415         """
 416         Verify captured packets with ICMP errors on outside network
 417
 418         :param capture: Captured packets
 419         :param src_ip: Translated IP address or IP address of VPP
 420                        (Default use global NAT address)
 421         :param icmp_type: Type of error ICMP packet
 422                           we are expecting (Default 11)
 423         """
 424         if src_ip is None:
 425             src_ip = self.nat_addr
 426         for packet in capture:
 427             try:
 428                 self.assertEqual(packet[IP].src, src_ip)
 429                 self.assertEqual(packet.haslayer(ICMP), 1)
 430                 icmp = packet[ICMP]
 431                 self.assertEqual(icmp.type, icmp_type)
 432                 self.assertTrue(icmp.haslayer(IPerror))
 433                 inner_ip = icmp[IPerror]
 434                 if inner_ip.haslayer(TCPerror):
 435                     self.assertEqual(inner_ip[TCPerror].dport,
 436                                      self.tcp_port_out)
 437                 elif inner_ip.haslayer(UDPerror):
 438                     self.assertEqual(inner_ip[UDPerror].dport,
 439                                      self.udp_port_out)
 440                 else:
 441                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
 442             except:
 443                 self.logger.error(ppp("Unexpected or invalid packet "
 444                                       "(outside network):", packet))
 445                 raise
 446
 447     def verify_capture_in_with_icmp_errors(self, capture, in_if, icmp_type=11):
 448         """
 449         Verify captured packets with ICMP errors on inside network
 450
 451         :param capture: Captured packets
 452         :param in_if: Inside interface
 453         :param icmp_type: Type of error ICMP packet
 454                           we are expecting (Default 11)
 455         """
 456         for packet in capture:
 457             try:
 458                 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
 459                 self.assertEqual(packet.haslayer(ICMP), 1)
 460                 icmp = packet[ICMP]
 461                 self.assertEqual(icmp.type, icmp_type)
 462                 self.assertTrue(icmp.haslayer(IPerror))
 463                 inner_ip = icmp[IPerror]
 464                 if inner_ip.haslayer(TCPerror):
 465                     self.assertEqual(inner_ip[TCPerror].sport,
 466                                      self.tcp_port_in)
 467                 elif inner_ip.haslayer(UDPerror):
 468                     self.assertEqual(inner_ip[UDPerror].sport,
 469                                      self.udp_port_in)
 470                 else:
 471                     self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
 472             except:
 473                 self.logger.error(ppp("Unexpected or invalid packet "
 474                                       "(inside network):", packet))
 475                 raise
 476
 477     def create_stream_frag(self, src_if, dst, sport, dport, data,
 478                            proto=IP_PROTOS.tcp, echo_reply=False):
 479         """
 480         Create fragmented packet stream
 481
 482         :param src_if: Source interface
 483         :param dst: Destination IPv4 address
 484         :param sport: Source port
 485         :param dport: Destination port
 486         :param data: Payload data
 487         :param proto: protocol (TCP, UDP, ICMP)
 488         :param echo_reply: use echo_reply if protocol is ICMP
 489         :returns: Fragments
 490         """
 491         if proto == IP_PROTOS.tcp:
 492             p = (IP(src=src_if.remote_ip4, dst=dst) /
 493                  TCP(sport=sport, dport=dport) /
 494                  Raw(data))
 495             p = p.__class__(scapy.compat.raw(p))
 496             chksum = p[TCP].chksum
 497             proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
 498         elif proto == IP_PROTOS.udp:
 499             proto_header = UDP(sport=sport, dport=dport)
 500         elif proto == IP_PROTOS.icmp:
 501             if not echo_reply:
 502                 proto_header = ICMP(id=sport, type='echo-request')
 503             else:
 504                 proto_header = ICMP(id=sport, type='echo-reply')
 505         else:
 506             raise Exception("Unsupported protocol")
 507         id = random.randint(0, 65535)
 508         pkts = []
 509         if proto == IP_PROTOS.tcp:
 510             raw = Raw(data[0:4])
 511         else:
 512             raw = Raw(data[0:16])
 513         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 514              IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) /
 515              proto_header /
 516              raw)
 517         pkts.append(p)
 518         if proto == IP_PROTOS.tcp:
 519             raw = Raw(data[4:20])
 520         else:
 521             raw = Raw(data[16:32])
 522         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 523              IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id,
 524                 proto=proto) /
 525              raw)
 526         pkts.append(p)
 527         if proto == IP_PROTOS.tcp:
 528             raw = Raw(data[20:])
 529         else:
 530             raw = Raw(data[32:])
 531         p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
 532              IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto,
 533                 id=id) /
 534              raw)
 535         pkts.append(p)
 536         return pkts
 537
 538     def reass_frags_and_verify(self, frags, src, dst):
 539         """
 540         Reassemble and verify fragmented packet
 541
 542         :param frags: Captured fragments
 543         :param src: Source IPv4 address to verify
 544         :param dst: Destination IPv4 address to verify
 545
 546         :returns: Reassembled IPv4 packet
 547         """
 548         buffer = BytesIO()
 549         for p in frags:
 550             self.assertEqual(p[IP].src, src)
 551             self.assertEqual(p[IP].dst, dst)
 552             self.assert_ip_checksum_valid(p)
 553             buffer.seek(p[IP].frag * 8)
 554             buffer.write(bytes(p[IP].payload))
 555         ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
 556                 proto=frags[0][IP].proto)
 557         if ip.proto == IP_PROTOS.tcp:
 558             p = (ip / TCP(buffer.getvalue()))
 559             self.logger.debug(ppp("Reassembled:", p))
 560             self.assert_tcp_checksum_valid(p)
 561         elif ip.proto == IP_PROTOS.udp:
 562             p = (ip / UDP(buffer.getvalue()[:8]) /
 563                  Raw(buffer.getvalue()[8:]))
 564         elif ip.proto == IP_PROTOS.icmp:
 565             p = (ip / ICMP(buffer.getvalue()))
 566         return p
 567
 568     def verify_ipfix_nat44_ses(self, data):
 569         """
 570         Verify IPFIX NAT44EI session create/delete event
 571
 572         :param data: Decoded IPFIX data records
 573         """
 574         nat44_ses_create_num = 0
 575         nat44_ses_delete_num = 0
 576         self.assertEqual(6, len(data))
 577         for record in data:
 578             # natEvent
 579             self.assertIn(scapy.compat.orb(record[230]), [4, 5])
 580             if scapy.compat.orb(record[230]) == 4:
 581                 nat44_ses_create_num += 1
 582             else:
 583                 nat44_ses_delete_num += 1
 584             # sourceIPv4Address
 585             self.assertEqual(self.pg0.remote_ip4,
 586                              str(ipaddress.IPv4Address(record[8])))
 587             # postNATSourceIPv4Address
 588             self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
 589                              record[225])
 590             # ingressVRFID
 591             self.assertEqual(struct.pack("!I", 0), record[234])
 592             # protocolIdentifier/sourceTransportPort
 593             # /postNAPTSourceTransportPort
 594             if IP_PROTOS.icmp == scapy.compat.orb(record[4]):
 595                 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
 596                 self.assertEqual(struct.pack("!H", self.icmp_id_out),
 597                                  record[227])
 598             elif IP_PROTOS.tcp == scapy.compat.orb(record[4]):
 599                 self.assertEqual(struct.pack("!H", self.tcp_port_in),
 600                                  record[7])
 601                 self.assertEqual(struct.pack("!H", self.tcp_port_out),
 602                                  record[227])
 603             elif IP_PROTOS.udp == scapy.compat.orb(record[4]):
 604                 self.assertEqual(struct.pack("!H", self.udp_port_in),
 605                                  record[7])
 606                 self.assertEqual(struct.pack("!H", self.udp_port_out),
 607                                  record[227])
 608             else:
 609                 self.fail("Invalid protocol")
 610         self.assertEqual(3, nat44_ses_create_num)
 611         self.assertEqual(3, nat44_ses_delete_num)
 612
 613     def verify_ipfix_addr_exhausted(self, data):
 614         self.assertEqual(1, len(data))
 615         record = data[0]
 616         # natEvent
 617         self.assertEqual(scapy.compat.orb(record[230]), 3)
 618         # natPoolID
 619         self.assertEqual(struct.pack("!I", 0), record[283])
 620
 621     def verify_ipfix_max_sessions(self, data, limit):
 622         self.assertEqual(1, len(data))
 623         record = data[0]
 624         # natEvent
 625         self.assertEqual(scapy.compat.orb(record[230]), 13)
 626         # natQuotaExceededEvent
 627         self.assertEqual(struct.pack("I", 1), record[466])
 628         # maxSessionEntries
 629         self.assertEqual(struct.pack("I", limit), record[471])
 630
 631     def verify_no_nat44_user(self):
 632         """ Verify that there is no NAT44EI user """
 633         users = self.vapi.nat44_ei_user_dump()
 634         self.assertEqual(len(users), 0)
 635         users = self.statistics.get_counter('/nat44-ei/total-users')
 636         self.assertEqual(users[0][0], 0)
 637         sessions = self.statistics.get_counter('/nat44-ei/total-sessions')
 638         self.assertEqual(sessions[0][0], 0)
 639
 640     def verify_syslog_apmap(self, data, is_add=True):
 641         message = data.decode('utf-8')
 642         try:
 643             message = SyslogMessage.parse(message)
 644         except ParseError as e:
 645             self.logger.error(e)
 646             raise
 647         else:
 648             self.assertEqual(message.severity, SyslogSeverity.info)
 649             self.assertEqual(message.appname, 'NAT')
 650             self.assertEqual(message.msgid, 'APMADD' if is_add else 'APMDEL')
 651             sd_params = message.sd.get('napmap')
 652             self.assertTrue(sd_params is not None)
 653             self.assertEqual(sd_params.get('IATYP'), 'IPv4')
 654             self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4)
 655             self.assertEqual(sd_params.get('ISPORT'), "%d" % self.tcp_port_in)
 656             self.assertEqual(sd_params.get('XATYP'), 'IPv4')
 657             self.assertEqual(sd_params.get('XSADDR'), self.nat_addr)
 658             self.assertEqual(sd_params.get('XSPORT'), "%d" % self.tcp_port_out)
 659             self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp)
 660             self.assertTrue(sd_params.get('SSUBIX') is not None)
 661             self.assertEqual(sd_params.get('SVLAN'), '0')
 662
 663     def verify_mss_value(self, pkt, mss):
 664         if not pkt.haslayer(IP) or not pkt.haslayer(TCP):
 665             raise TypeError("Not a TCP/IP packet")
 666
 667         for option in pkt[TCP].options:
 668             if option[0] == 'MSS':
 669                 self.assertEqual(option[1], mss)
 670                 self.assert_tcp_checksum_valid(pkt)
 671
 672     @staticmethod
 673     def proto2layer(proto):
 674         if proto == IP_PROTOS.tcp:
 675             return TCP
 676         elif proto == IP_PROTOS.udp:
 677             return UDP
 678         elif proto == IP_PROTOS.icmp:
 679             return ICMP
 680         else:
 681             raise Exception("Unsupported protocol")
 682
 683     def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
 684                       ignore_port=False):
 685         layer = self.proto2layer(proto)
 686
 687         if proto == IP_PROTOS.tcp:
 688             data = b"A" * 4 + b"B" * 16 + b"C" * 3
 689         else:
 690             data = b"A" * 16 + b"B" * 16 + b"C" * 3
 691         self.port_in = random.randint(1025, 65535)
 692
 693         # in2out
 694         pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
 695                                        self.port_in, 20, data, proto)
 696         self.pg0.add_stream(pkts)
 697         self.pg_enable_capture(self.pg_interfaces)
 698         self.pg_start()
 699         frags = self.pg1.get_capture(len(pkts))
 700         if not dont_translate:
 701             p = self.reass_frags_and_verify(frags,
 702                                             self.nat_addr,
 703                                             self.pg1.remote_ip4)
 704         else:
 705             p = self.reass_frags_and_verify(frags,
 706                                             self.pg0.remote_ip4,
 707                                             self.pg1.remote_ip4)
 708         if proto != IP_PROTOS.icmp:
 709             if not dont_translate:
 710                 self.assertEqual(p[layer].dport, 20)
 711                 if not ignore_port:
 712                     self.assertNotEqual(p[layer].sport, self.port_in)
 713             else:
 714                 self.assertEqual(p[layer].sport, self.port_in)
 715         else:
 716             if not ignore_port:
 717                 if not dont_translate:
 718                     self.assertNotEqual(p[layer].id, self.port_in)
 719                 else:
 720                     self.assertEqual(p[layer].id, self.port_in)
 721         self.assertEqual(data, p[Raw].load)
 722
 723         # out2in
 724         if not dont_translate:
 725             dst_addr = self.nat_addr
 726         else:
 727             dst_addr = self.pg0.remote_ip4
 728         if proto != IP_PROTOS.icmp:
 729             sport = 20
 730             dport = p[layer].sport
 731         else:
 732             sport = p[layer].id
 733             dport = 0
 734         pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, data,
 735                                        proto, echo_reply=True)
 736         self.pg1.add_stream(pkts)
 737         self.pg_enable_capture(self.pg_interfaces)
 738         self.pg_start()
 739         frags = self.pg0.get_capture(len(pkts))
 740         p = self.reass_frags_and_verify(frags,
 741                                         self.pg1.remote_ip4,
 742                                         self.pg0.remote_ip4)
 743         if proto != IP_PROTOS.icmp:
 744             self.assertEqual(p[layer].sport, 20)
 745             self.assertEqual(p[layer].dport, self.port_in)
 746         else:
 747             self.assertEqual(p[layer].id, self.port_in)
 748         self.assertEqual(data, p[Raw].load)
 749
 750     def reass_hairpinning(self, server_addr, server_in_port, server_out_port,
 751                           host_in_port, proto=IP_PROTOS.tcp,
 752                           ignore_port=False):
 753
 754         layer = self.proto2layer(proto)
 755
 756         if proto == IP_PROTOS.tcp:
 757             data = b"A" * 4 + b"B" * 16 + b"C" * 3
 758         else:
 759             data = b"A" * 16 + b"B" * 16 + b"C" * 3
 760
 761         # send packet from host to server
 762         pkts = self.create_stream_frag(self.pg0,
 763                                        self.nat_addr,
 764                                        host_in_port,
 765                                        server_out_port,
 766                                        data,
 767                                        proto)
 768         self.pg0.add_stream(pkts)
 769         self.pg_enable_capture(self.pg_interfaces)
 770         self.pg_start()
 771         frags = self.pg0.get_capture(len(pkts))
 772         p = self.reass_frags_and_verify(frags,
 773                                         self.nat_addr,
 774                                         server_addr)
 775         if proto != IP_PROTOS.icmp:
 776             if not ignore_port:
 777                 self.assertNotEqual(p[layer].sport, host_in_port)
 778             self.assertEqual(p[layer].dport, server_in_port)
 779         else:
 780             if not ignore_port:
 781                 self.assertNotEqual(p[layer].id, host_in_port)
 782         self.assertEqual(data, p[Raw].load)
 783
 784     def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
 785                           ignore_port=False):
 786         layer = self.proto2layer(proto)
 787
 788         if proto == IP_PROTOS.tcp:
 789             data = b"A" * 4 + b"B" * 16 + b"C" * 3
 790         else:
 791             data = b"A" * 16 + b"B" * 16 + b"C" * 3
 792         self.port_in = random.randint(1025, 65535)
 793
 794         for i in range(2):
 795             # in2out
 796             pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
 797                                            self.port_in, 20, data, proto)
 798             pkts.reverse()
 799             self.pg0.add_stream(pkts)
 800             self.pg_enable_capture(self.pg_interfaces)
 801             self.pg_start()
 802             frags = self.pg1.get_capture(len(pkts))
 803             if not dont_translate:
 804                 p = self.reass_frags_and_verify(frags,
 805                                                 self.nat_addr,
 806                                                 self.pg1.remote_ip4)
 807             else:
 808                 p = self.reass_frags_and_verify(frags,
 809                                                 self.pg0.remote_ip4,
 810                                                 self.pg1.remote_ip4)
 811             if proto != IP_PROTOS.icmp:
 812                 if not dont_translate:
 813                     self.assertEqual(p[layer].dport, 20)
 814                     if not ignore_port:
 815                         self.assertNotEqual(p[layer].sport, self.port_in)
 816                 else:
 817                     self.assertEqual(p[layer].sport, self.port_in)
 818             else:
 819                 if not ignore_port:
 820                     if not dont_translate:
 821                         self.assertNotEqual(p[layer].id, self.port_in)
 822                     else:
 823                         self.assertEqual(p[layer].id, self.port_in)
 824             self.assertEqual(data, p[Raw].load)
 825
 826             # out2in
 827             if not dont_translate:
 828                 dst_addr = self.nat_addr
 829             else:
 830                 dst_addr = self.pg0.remote_ip4
 831             if proto != IP_PROTOS.icmp:
 832                 sport = 20
 833                 dport = p[layer].sport
 834             else:
 835                 sport = p[layer].id
 836                 dport = 0
 837             pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport,
 838                                            data, proto, echo_reply=True)
 839             pkts.reverse()
 840             self.pg1.add_stream(pkts)
 841             self.pg_enable_capture(self.pg_interfaces)
 842             self.pg_start()
 843             frags = self.pg0.get_capture(len(pkts))
 844             p = self.reass_frags_and_verify(frags,
 845                                             self.pg1.remote_ip4,
 846                                             self.pg0.remote_ip4)
 847             if proto != IP_PROTOS.icmp:
 848                 self.assertEqual(p[layer].sport, 20)
 849                 self.assertEqual(p[layer].dport, self.port_in)
 850             else:
 851                 self.assertEqual(p[layer].id, self.port_in)
 852             self.assertEqual(data, p[Raw].load)
 853
 854
 855 @tag_fixme_vpp_workers
 856 class TestNAT44EI(MethodHolder):
 857     """ NAT44EI Test Cases """
 858
 859     max_translations = 10240
 860     max_users = 10240
 861
 862     @classmethod
 863     def setUpClass(cls):
 864         super(TestNAT44EI, cls).setUpClass()
 865         cls.vapi.cli("set log class nat44-ei level debug")
 866
 867         cls.tcp_port_in = 6303
 868         cls.tcp_port_out = 6303
 869         cls.udp_port_in = 6304
 870         cls.udp_port_out = 6304
 871         cls.icmp_id_in = 6305
 872         cls.icmp_id_out = 6305
 873         cls.nat_addr = '10.0.0.3'
 874         cls.ipfix_src_port = 4739
 875         cls.ipfix_domain_id = 1
 876         cls.tcp_external_port = 80
 877         cls.udp_external_port = 69
 878
 879         cls.create_pg_interfaces(range(10))
 880         cls.interfaces = list(cls.pg_interfaces[0:4])
 881
 882         for i in cls.interfaces:
 883             i.admin_up()
 884             i.config_ip4()
 885             i.resolve_arp()
 886
 887         cls.pg0.generate_remote_hosts(3)
 888         cls.pg0.configure_ipv4_neighbors()
 889
 890         cls.pg1.generate_remote_hosts(1)
 891         cls.pg1.configure_ipv4_neighbors()
 892
 893         cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
 894         cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
 895         cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
 896
 897         cls.pg4._local_ip4 = "172.16.255.1"
 898         cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
 899         cls.pg4.set_table_ip4(10)
 900         cls.pg5._local_ip4 = "172.17.255.3"
 901         cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
 902         cls.pg5.set_table_ip4(10)
 903         cls.pg6._local_ip4 = "172.16.255.1"
 904         cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
 905         cls.pg6.set_table_ip4(20)
 906         for i in cls.overlapping_interfaces:
 907             i.config_ip4()
 908             i.admin_up()
 909             i.resolve_arp()
 910
 911         cls.pg7.admin_up()
 912         cls.pg8.admin_up()
 913
 914         cls.pg9.generate_remote_hosts(2)
 915         cls.pg9.config_ip4()
 916         cls.vapi.sw_interface_add_del_address(
 917             sw_if_index=cls.pg9.sw_if_index,
 918             prefix="10.0.0.1/24")
 919
 920         cls.pg9.admin_up()
 921         cls.pg9.resolve_arp()
 922         cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
 923         cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
 924         cls.pg9.resolve_arp()
 925
 926     def plugin_enable(self):
 927         self.vapi.nat44_ei_plugin_enable_disable(
 928             sessions=self.max_translations,
 929             users=self.max_users, enable=1)
 930
 931     def setUp(self):
 932         super(TestNAT44EI, self).setUp()
 933         self.plugin_enable()
 934
 935     def tearDown(self):
 936         super(TestNAT44EI, self).tearDown()
 937         if not self.vpp_dead:
 938             self.vapi.nat44_ei_ipfix_enable_disable(
 939                 domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port,
 940                 enable=0)
 941             self.ipfix_src_port = 4739
 942             self.ipfix_domain_id = 1
 943
 944             self.vapi.nat44_ei_plugin_enable_disable(enable=0)
 945             self.vapi.cli("clear logging")
 946
 947     def test_clear_sessions(self):
 948         """ NAT44EI session clearing test """
 949
 950         self.nat44_add_address(self.nat_addr)
 951         flags = self.config_flags.NAT44_EI_IF_INSIDE
 952         self.vapi.nat44_ei_interface_add_del_feature(
 953             sw_if_index=self.pg0.sw_if_index,
 954             flags=flags, is_add=1)
 955         self.vapi.nat44_ei_interface_add_del_feature(
 956             sw_if_index=self.pg1.sw_if_index,
 957             is_add=1)
 958
 959         pkts = self.create_stream_in(self.pg0, self.pg1)
 960         self.pg0.add_stream(pkts)
 961         self.pg_enable_capture(self.pg_interfaces)
 962         self.pg_start()
 963         capture = self.pg1.get_capture(len(pkts))
 964         self.verify_capture_out(capture)
 965
 966         sessions = self.statistics.get_counter('/nat44-ei/total-sessions')
 967         self.assertTrue(sessions[0][0] > 0)
 968         self.logger.info("sessions before clearing: %s" % sessions[0][0])
 969
 970         self.vapi.cli("clear nat44 ei sessions")
 971
 972         sessions = self.statistics.get_counter('/nat44-ei/total-sessions')
 973         self.assertEqual(sessions[0][0], 0)
 974         self.logger.info("sessions after clearing: %s" % sessions[0][0])
 975
 976     def test_dynamic(self):
 977         """ NAT44EI dynamic translation test """
 978         self.nat44_add_address(self.nat_addr)
 979         flags = self.config_flags.NAT44_EI_IF_INSIDE
 980         self.vapi.nat44_ei_interface_add_del_feature(
 981             sw_if_index=self.pg0.sw_if_index,
 982             flags=flags, is_add=1)
 983         self.vapi.nat44_ei_interface_add_del_feature(
 984             sw_if_index=self.pg1.sw_if_index,
 985             is_add=1)
 986
 987         # in2out
 988         tcpn = self.statistics.get_counter('/nat44-ei/in2out/slowpath/tcp')[0]
 989         udpn = self.statistics.get_counter('/nat44-ei/in2out/slowpath/udp')[0]
 990         icmpn = self.statistics.get_counter(
 991             '/nat44-ei/in2out/slowpath/icmp')[0]
 992         drops = self.statistics.get_counter(
 993             '/nat44-ei/in2out/slowpath/drops')[0]
 994
 995         pkts = self.create_stream_in(self.pg0, self.pg1)
 996         self.pg0.add_stream(pkts)
 997         self.pg_enable_capture(self.pg_interfaces)
 998         self.pg_start()
 999         capture = self.pg1.get_capture(len(pkts))
1000         self.verify_capture_out(capture)
1001
1002         if_idx = self.pg0.sw_if_index
1003         cnt = self.statistics.get_counter('/nat44-ei/in2out/slowpath/tcp')[0]
1004         self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
1005         cnt = self.statistics.get_counter('/nat44-ei/in2out/slowpath/udp')[0]
1006         self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
1007         cnt = self.statistics.get_counter('/nat44-ei/in2out/slowpath/icmp')[0]
1008         self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
1009         cnt = self.statistics.get_counter('/nat44-ei/in2out/slowpath/drops')[0]
1010         self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
1011
1012         # out2in
1013         tcpn = self.statistics.get_counter('/nat44-ei/out2in/slowpath/tcp')[0]
1014         udpn = self.statistics.get_counter('/nat44-ei/out2in/slowpath/udp')[0]
1015         icmpn = self.statistics.get_counter(
1016             '/nat44-ei/out2in/slowpath/icmp')[0]
1017         drops = self.statistics.get_counter(
1018             '/nat44-ei/out2in/slowpath/drops')[0]
1019
1020         pkts = self.create_stream_out(self.pg1)
1021         self.pg1.add_stream(pkts)
1022         self.pg_enable_capture(self.pg_interfaces)
1023         self.pg_start()
1024         capture = self.pg0.get_capture(len(pkts))
1025         self.verify_capture_in(capture, self.pg0)
1026
1027         if_idx = self.pg1.sw_if_index
1028         cnt = self.statistics.get_counter('/nat44-ei/out2in/slowpath/tcp')[0]
1029         self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
1030         cnt = self.statistics.get_counter('/nat44-ei/out2in/slowpath/udp')[0]
1031         self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
1032         cnt = self.statistics.get_counter('/nat44-ei/out2in/slowpath/icmp')[0]
1033         self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
1034         cnt = self.statistics.get_counter('/nat44-ei/out2in/slowpath/drops')[0]
1035         self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
1036
1037         users = self.statistics.get_counter('/nat44-ei/total-users')
1038         self.assertEqual(users[0][0], 1)
1039         sessions = self.statistics.get_counter('/nat44-ei/total-sessions')
1040         self.assertEqual(sessions[0][0], 3)
1041
1042     def test_dynamic_icmp_errors_in2out_ttl_1(self):
1043         """ NAT44EI handling of client packets with TTL=1 """
1044
1045         self.nat44_add_address(self.nat_addr)
1046         flags = self.config_flags.NAT44_EI_IF_INSIDE
1047         self.vapi.nat44_ei_interface_add_del_feature(
1048             sw_if_index=self.pg0.sw_if_index,
1049             flags=flags, is_add=1)
1050         self.vapi.nat44_ei_interface_add_del_feature(
1051             sw_if_index=self.pg1.sw_if_index,
1052             is_add=1)
1053
1054         # Client side - generate traffic
1055         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
1056         self.pg0.add_stream(pkts)
1057         self.pg_enable_capture(self.pg_interfaces)
1058         self.pg_start()
1059
1060         # Client side - verify ICMP type 11 packets
1061         capture = self.pg0.get_capture(len(pkts))
1062         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1063
1064     def test_dynamic_icmp_errors_out2in_ttl_1(self):
1065         """ NAT44EI handling of server packets with TTL=1 """
1066
1067         self.nat44_add_address(self.nat_addr)
1068         flags = self.config_flags.NAT44_EI_IF_INSIDE
1069         self.vapi.nat44_ei_interface_add_del_feature(
1070             sw_if_index=self.pg0.sw_if_index,
1071             flags=flags, is_add=1)
1072         self.vapi.nat44_ei_interface_add_del_feature(
1073             sw_if_index=self.pg1.sw_if_index,
1074             is_add=1)
1075
1076         # Client side - create sessions
1077         pkts = self.create_stream_in(self.pg0, self.pg1)
1078         self.pg0.add_stream(pkts)
1079         self.pg_enable_capture(self.pg_interfaces)
1080         self.pg_start()
1081
1082         # Server side - generate traffic
1083         capture = self.pg1.get_capture(len(pkts))
1084         self.verify_capture_out(capture)
1085         pkts = self.create_stream_out(self.pg1, ttl=1)
1086         self.pg1.add_stream(pkts)
1087         self.pg_enable_capture(self.pg_interfaces)
1088         self.pg_start()
1089
1090         # Server side - verify ICMP type 11 packets
1091         capture = self.pg1.get_capture(len(pkts))
1092         self.verify_capture_out_with_icmp_errors(capture,
1093                                                  src_ip=self.pg1.local_ip4)
1094
1095     def test_dynamic_icmp_errors_in2out_ttl_2(self):
1096         """ NAT44EI handling of error responses to client packets with TTL=2
1097         """
1098
1099         self.nat44_add_address(self.nat_addr)
1100         flags = self.config_flags.NAT44_EI_IF_INSIDE
1101         self.vapi.nat44_ei_interface_add_del_feature(
1102             sw_if_index=self.pg0.sw_if_index,
1103             flags=flags, is_add=1)
1104         self.vapi.nat44_ei_interface_add_del_feature(
1105             sw_if_index=self.pg1.sw_if_index,
1106             is_add=1)
1107
1108         # Client side - generate traffic
1109         pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
1110         self.pg0.add_stream(pkts)
1111         self.pg_enable_capture(self.pg_interfaces)
1112         self.pg_start()
1113
1114         # Server side - simulate ICMP type 11 response
1115         capture = self.pg1.get_capture(len(pkts))
1116         pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1117                 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1118                 ICMP(type=11) / packet[IP] for packet in capture]
1119         self.pg1.add_stream(pkts)
1120         self.pg_enable_capture(self.pg_interfaces)
1121         self.pg_start()
1122
1123         # Client side - verify ICMP type 11 packets
1124         capture = self.pg0.get_capture(len(pkts))
1125         self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1126
1127     def test_dynamic_icmp_errors_out2in_ttl_2(self):
1128         """ NAT44EI handling of error responses to server packets with TTL=2
1129         """
1130
1131         self.nat44_add_address(self.nat_addr)
1132         flags = self.config_flags.NAT44_EI_IF_INSIDE
1133         self.vapi.nat44_ei_interface_add_del_feature(
1134             sw_if_index=self.pg0.sw_if_index,
1135             flags=flags, is_add=1)
1136         self.vapi.nat44_ei_interface_add_del_feature(
1137             sw_if_index=self.pg1.sw_if_index,
1138             is_add=1)
1139
1140         # Client side - create sessions
1141         pkts = self.create_stream_in(self.pg0, self.pg1)
1142         self.pg0.add_stream(pkts)
1143         self.pg_enable_capture(self.pg_interfaces)
1144         self.pg_start()
1145
1146         # Server side - generate traffic
1147         capture = self.pg1.get_capture(len(pkts))
1148         self.verify_capture_out(capture)
1149         pkts = self.create_stream_out(self.pg1, ttl=2)
1150         self.pg1.add_stream(pkts)
1151         self.pg_enable_capture(self.pg_interfaces)
1152         self.pg_start()
1153
1154         # Client side - simulate ICMP type 11 response
1155         capture = self.pg0.get_capture(len(pkts))
1156         pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1157                 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1158                 ICMP(type=11) / packet[IP] for packet in capture]
1159         self.pg0.add_stream(pkts)
1160         self.pg_enable_capture(self.pg_interfaces)
1161         self.pg_start()
1162
1163         # Server side - verify ICMP type 11 packets
1164         capture = self.pg1.get_capture(len(pkts))
1165         self.verify_capture_out_with_icmp_errors(capture)
1166
1167     def test_ping_out_interface_from_outside(self):
1168         """ NAT44EI ping out interface from outside network """
1169
1170         self.nat44_add_address(self.nat_addr)
1171         flags = self.config_flags.NAT44_EI_IF_INSIDE
1172         self.vapi.nat44_ei_interface_add_del_feature(
1173             sw_if_index=self.pg0.sw_if_index,
1174             flags=flags, is_add=1)
1175         self.vapi.nat44_ei_interface_add_del_feature(
1176             sw_if_index=self.pg1.sw_if_index,
1177             is_add=1)
1178
1179         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1180              IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
1181              ICMP(id=self.icmp_id_out, type='echo-request'))
1182         pkts = [p]
1183         self.pg1.add_stream(pkts)
1184         self.pg_enable_capture(self.pg_interfaces)
1185         self.pg_start()
1186         capture = self.pg1.get_capture(len(pkts))
1187         packet = capture[0]
1188         try:
1189             self.assertEqual(packet[IP].src, self.pg1.local_ip4)
1190             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
1191             self.assertEqual(packet[ICMP].id, self.icmp_id_in)
1192             self.assertEqual(packet[ICMP].type, 0)  # echo reply
1193         except:
1194             self.logger.error(ppp("Unexpected or invalid packet "
1195                                   "(outside network):", packet))
1196             raise
1197
1198     def test_ping_internal_host_from_outside(self):
1199         """ NAT44EI ping internal host from outside network """
1200
1201         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
1202         flags = self.config_flags.NAT44_EI_IF_INSIDE
1203         self.vapi.nat44_ei_interface_add_del_feature(
1204             sw_if_index=self.pg0.sw_if_index,
1205             flags=flags, is_add=1)
1206         self.vapi.nat44_ei_interface_add_del_feature(
1207             sw_if_index=self.pg1.sw_if_index,
1208             is_add=1)
1209
1210         # out2in
1211         pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1212                IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
1213                ICMP(id=self.icmp_id_out, type='echo-request'))
1214         self.pg1.add_stream(pkt)
1215         self.pg_enable_capture(self.pg_interfaces)
1216         self.pg_start()
1217         capture = self.pg0.get_capture(1)
1218         self.verify_capture_in(capture, self.pg0)
1219         self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1220
1221         # in2out
1222         pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1223                IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
1224                ICMP(id=self.icmp_id_in, type='echo-reply'))
1225         self.pg0.add_stream(pkt)
1226         self.pg_enable_capture(self.pg_interfaces)
1227         self.pg_start()
1228         capture = self.pg1.get_capture(1)
1229         self.verify_capture_out(capture, same_port=True)
1230         self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1231
1232     def test_forwarding(self):
1233         """ NAT44EI forwarding test """
1234
1235         flags = self.config_flags.NAT44_EI_IF_INSIDE
1236         self.vapi.nat44_ei_interface_add_del_feature(
1237             sw_if_index=self.pg0.sw_if_index,
1238             flags=flags, is_add=1)
1239         self.vapi.nat44_ei_interface_add_del_feature(
1240             sw_if_index=self.pg1.sw_if_index,
1241             is_add=1)
1242         self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
1243
1244         real_ip = self.pg0.remote_ip4
1245         alias_ip = self.nat_addr
1246         flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1247         self.vapi.nat44_ei_add_del_static_mapping(
1248             is_add=1, local_ip_address=real_ip,
1249             external_ip_address=alias_ip,
1250             external_sw_if_index=0xFFFFFFFF,
1251             flags=flags)
1252
1253         try:
1254             # static mapping match
1255
1256             pkts = self.create_stream_out(self.pg1)
1257             self.pg1.add_stream(pkts)
1258             self.pg_enable_capture(self.pg_interfaces)
1259             self.pg_start()
1260             capture = self.pg0.get_capture(len(pkts))
1261             self.verify_capture_in(capture, self.pg0)
1262
1263             pkts = self.create_stream_in(self.pg0, self.pg1)
1264             self.pg0.add_stream(pkts)
1265             self.pg_enable_capture(self.pg_interfaces)
1266             self.pg_start()
1267             capture = self.pg1.get_capture(len(pkts))
1268             self.verify_capture_out(capture, same_port=True)
1269
1270             # no static mapping match
1271
1272             host0 = self.pg0.remote_hosts[0]
1273             self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1274             try:
1275                 pkts = self.create_stream_out(self.pg1,
1276                                               dst_ip=self.pg0.remote_ip4,
1277                                               use_inside_ports=True)
1278                 self.pg1.add_stream(pkts)
1279                 self.pg_enable_capture(self.pg_interfaces)
1280                 self.pg_start()
1281                 capture = self.pg0.get_capture(len(pkts))
1282                 self.verify_capture_in(capture, self.pg0)
1283
1284                 pkts = self.create_stream_in(self.pg0, self.pg1)
1285                 self.pg0.add_stream(pkts)
1286                 self.pg_enable_capture(self.pg_interfaces)
1287                 self.pg_start()
1288                 capture = self.pg1.get_capture(len(pkts))
1289                 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
1290                                         same_port=True)
1291             finally:
1292                 self.pg0.remote_hosts[0] = host0
1293
1294         finally:
1295             self.vapi.nat44_ei_forwarding_enable_disable(enable=0)
1296             flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1297             self.vapi.nat44_ei_add_del_static_mapping(
1298                 is_add=0,
1299                 local_ip_address=real_ip,
1300                 external_ip_address=alias_ip,
1301                 external_sw_if_index=0xFFFFFFFF,
1302                 flags=flags)
1303
1304     def test_static_in(self):
1305         """ NAT44EI 1:1 NAT initialized from inside network """
1306
1307         nat_ip = "10.0.0.10"
1308         self.tcp_port_out = 6303
1309         self.udp_port_out = 6304
1310         self.icmp_id_out = 6305
1311
1312         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1313         flags = self.config_flags.NAT44_EI_IF_INSIDE
1314         self.vapi.nat44_ei_interface_add_del_feature(
1315             sw_if_index=self.pg0.sw_if_index,
1316             flags=flags, is_add=1)
1317         self.vapi.nat44_ei_interface_add_del_feature(
1318             sw_if_index=self.pg1.sw_if_index,
1319             is_add=1)
1320         sm = self.vapi.nat44_ei_static_mapping_dump()
1321         self.assertEqual(len(sm), 1)
1322         self.assertEqual(sm[0].tag, '')
1323         self.assertEqual(sm[0].protocol, 0)
1324         self.assertEqual(sm[0].local_port, 0)
1325         self.assertEqual(sm[0].external_port, 0)
1326
1327         # in2out
1328         pkts = self.create_stream_in(self.pg0, self.pg1)
1329         self.pg0.add_stream(pkts)
1330         self.pg_enable_capture(self.pg_interfaces)
1331         self.pg_start()
1332         capture = self.pg1.get_capture(len(pkts))
1333         self.verify_capture_out(capture, nat_ip, True)
1334
1335         # out2in
1336         pkts = self.create_stream_out(self.pg1, nat_ip)
1337         self.pg1.add_stream(pkts)
1338         self.pg_enable_capture(self.pg_interfaces)
1339         self.pg_start()
1340         capture = self.pg0.get_capture(len(pkts))
1341         self.verify_capture_in(capture, self.pg0)
1342
1343     def test_static_out(self):
1344         """ NAT44EI 1:1 NAT initialized from outside network """
1345
1346         nat_ip = "10.0.0.20"
1347         self.tcp_port_out = 6303
1348         self.udp_port_out = 6304
1349         self.icmp_id_out = 6305
1350         tag = "testTAG"
1351
1352         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
1353         flags = self.config_flags.NAT44_EI_IF_INSIDE
1354         self.vapi.nat44_ei_interface_add_del_feature(
1355             sw_if_index=self.pg0.sw_if_index,
1356             flags=flags, is_add=1)
1357         self.vapi.nat44_ei_interface_add_del_feature(
1358             sw_if_index=self.pg1.sw_if_index,
1359             is_add=1)
1360         sm = self.vapi.nat44_ei_static_mapping_dump()
1361         self.assertEqual(len(sm), 1)
1362         self.assertEqual(sm[0].tag, tag)
1363
1364         # out2in
1365         pkts = self.create_stream_out(self.pg1, nat_ip)
1366         self.pg1.add_stream(pkts)
1367         self.pg_enable_capture(self.pg_interfaces)
1368         self.pg_start()
1369         capture = self.pg0.get_capture(len(pkts))
1370         self.verify_capture_in(capture, self.pg0)
1371
1372         # in2out
1373         pkts = self.create_stream_in(self.pg0, self.pg1)
1374         self.pg0.add_stream(pkts)
1375         self.pg_enable_capture(self.pg_interfaces)
1376         self.pg_start()
1377         capture = self.pg1.get_capture(len(pkts))
1378         self.verify_capture_out(capture, nat_ip, True)
1379
1380     def test_static_with_port_in(self):
1381         """ NAT44EI 1:1 NAPT initialized from inside network """
1382
1383         self.tcp_port_out = 3606
1384         self.udp_port_out = 3607
1385         self.icmp_id_out = 3608
1386
1387         self.nat44_add_address(self.nat_addr)
1388         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1389                                       self.tcp_port_in, self.tcp_port_out,
1390                                       proto=IP_PROTOS.tcp)
1391         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1392                                       self.udp_port_in, self.udp_port_out,
1393                                       proto=IP_PROTOS.udp)
1394         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1395                                       self.icmp_id_in, self.icmp_id_out,
1396                                       proto=IP_PROTOS.icmp)
1397         flags = self.config_flags.NAT44_EI_IF_INSIDE
1398         self.vapi.nat44_ei_interface_add_del_feature(
1399             sw_if_index=self.pg0.sw_if_index,
1400             flags=flags, is_add=1)
1401         self.vapi.nat44_ei_interface_add_del_feature(
1402             sw_if_index=self.pg1.sw_if_index,
1403             is_add=1)
1404
1405         # in2out
1406         pkts = self.create_stream_in(self.pg0, self.pg1)
1407         self.pg0.add_stream(pkts)
1408         self.pg_enable_capture(self.pg_interfaces)
1409         self.pg_start()
1410         capture = self.pg1.get_capture(len(pkts))
1411         self.verify_capture_out(capture)
1412
1413         # out2in
1414         pkts = self.create_stream_out(self.pg1)
1415         self.pg1.add_stream(pkts)
1416         self.pg_enable_capture(self.pg_interfaces)
1417         self.pg_start()
1418         capture = self.pg0.get_capture(len(pkts))
1419         self.verify_capture_in(capture, self.pg0)
1420
1421     def test_static_with_port_out(self):
1422         """ NAT44EI 1:1 NAPT initialized from outside network """
1423
1424         self.tcp_port_out = 30606
1425         self.udp_port_out = 30607
1426         self.icmp_id_out = 30608
1427
1428         self.nat44_add_address(self.nat_addr)
1429         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1430                                       self.tcp_port_in, self.tcp_port_out,
1431                                       proto=IP_PROTOS.tcp)
1432         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1433                                       self.udp_port_in, self.udp_port_out,
1434                                       proto=IP_PROTOS.udp)
1435         self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1436                                       self.icmp_id_in, self.icmp_id_out,
1437                                       proto=IP_PROTOS.icmp)
1438         flags = self.config_flags.NAT44_EI_IF_INSIDE
1439         self.vapi.nat44_ei_interface_add_del_feature(
1440             sw_if_index=self.pg0.sw_if_index,
1441             flags=flags, is_add=1)
1442         self.vapi.nat44_ei_interface_add_del_feature(
1443             sw_if_index=self.pg1.sw_if_index,
1444             is_add=1)
1445
1446         # out2in
1447         pkts = self.create_stream_out(self.pg1)
1448         self.pg1.add_stream(pkts)
1449         self.pg_enable_capture(self.pg_interfaces)
1450         self.pg_start()
1451         capture = self.pg0.get_capture(len(pkts))
1452         self.verify_capture_in(capture, self.pg0)
1453
1454         # in2out
1455         pkts = self.create_stream_in(self.pg0, self.pg1)
1456         self.pg0.add_stream(pkts)
1457         self.pg_enable_capture(self.pg_interfaces)
1458         self.pg_start()
1459         capture = self.pg1.get_capture(len(pkts))
1460         self.verify_capture_out(capture)
1461
1462     def test_static_vrf_aware(self):
1463         """ NAT44EI 1:1 NAT VRF awareness """
1464
1465         nat_ip1 = "10.0.0.30"
1466         nat_ip2 = "10.0.0.40"
1467         self.tcp_port_out = 6303
1468         self.udp_port_out = 6304
1469         self.icmp_id_out = 6305
1470
1471         self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
1472                                       vrf_id=10)
1473         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
1474                                       vrf_id=10)
1475         flags = self.config_flags.NAT44_EI_IF_INSIDE
1476         self.vapi.nat44_ei_interface_add_del_feature(
1477             sw_if_index=self.pg3.sw_if_index,
1478             is_add=1)
1479         self.vapi.nat44_ei_interface_add_del_feature(
1480             sw_if_index=self.pg0.sw_if_index,
1481             flags=flags, is_add=1)
1482         self.vapi.nat44_ei_interface_add_del_feature(
1483             sw_if_index=self.pg4.sw_if_index,
1484             flags=flags, is_add=1)
1485
1486         # inside interface VRF match NAT44EI static mapping VRF
1487         pkts = self.create_stream_in(self.pg4, self.pg3)
1488         self.pg4.add_stream(pkts)
1489         self.pg_enable_capture(self.pg_interfaces)
1490         self.pg_start()
1491         capture = self.pg3.get_capture(len(pkts))
1492         self.verify_capture_out(capture, nat_ip1, True)
1493
1494         # inside interface VRF don't match NAT44EI static mapping VRF (packets
1495         # are dropped)
1496         pkts = self.create_stream_in(self.pg0, self.pg3)
1497         self.pg0.add_stream(pkts)
1498         self.pg_enable_capture(self.pg_interfaces)
1499         self.pg_start()
1500         self.pg3.assert_nothing_captured()
1501
1502     def test_dynamic_to_static(self):
1503         """ NAT44EI Switch from dynamic translation to 1:1NAT """
1504         nat_ip = "10.0.0.10"
1505         self.tcp_port_out = 6303
1506         self.udp_port_out = 6304
1507         self.icmp_id_out = 6305
1508
1509         self.nat44_add_address(self.nat_addr)
1510         flags = self.config_flags.NAT44_EI_IF_INSIDE
1511         self.vapi.nat44_ei_interface_add_del_feature(
1512             sw_if_index=self.pg0.sw_if_index,
1513             flags=flags, is_add=1)
1514         self.vapi.nat44_ei_interface_add_del_feature(
1515             sw_if_index=self.pg1.sw_if_index,
1516             is_add=1)
1517
1518         # dynamic
1519         pkts = self.create_stream_in(self.pg0, self.pg1)
1520         self.pg0.add_stream(pkts)
1521         self.pg_enable_capture(self.pg_interfaces)
1522         self.pg_start()
1523         capture = self.pg1.get_capture(len(pkts))
1524         self.verify_capture_out(capture)
1525
1526         # 1:1NAT
1527         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1528         sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
1529         self.assertEqual(len(sessions), 0)
1530         pkts = self.create_stream_in(self.pg0, self.pg1)
1531         self.pg0.add_stream(pkts)
1532         self.pg_enable_capture(self.pg_interfaces)
1533         self.pg_start()
1534         capture = self.pg1.get_capture(len(pkts))
1535         self.verify_capture_out(capture, nat_ip, True)
1536
1537     def test_identity_nat(self):
1538         """ NAT44EI Identity NAT """
1539         flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1540         self.vapi.nat44_ei_add_del_identity_mapping(
1541             ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
1542             flags=flags, is_add=1)
1543         flags = self.config_flags.NAT44_EI_IF_INSIDE
1544         self.vapi.nat44_ei_interface_add_del_feature(
1545             sw_if_index=self.pg0.sw_if_index,
1546             flags=flags, is_add=1)
1547         self.vapi.nat44_ei_interface_add_del_feature(
1548             sw_if_index=self.pg1.sw_if_index,
1549             is_add=1)
1550
1551         p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1552              IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
1553              TCP(sport=12345, dport=56789))
1554         self.pg1.add_stream(p)
1555         self.pg_enable_capture(self.pg_interfaces)
1556         self.pg_start()
1557         capture = self.pg0.get_capture(1)
1558         p = capture[0]
1559         try:
1560             ip = p[IP]
1561             tcp = p[TCP]
1562             self.assertEqual(ip.dst, self.pg0.remote_ip4)
1563             self.assertEqual(ip.src, self.pg1.remote_ip4)
1564             self.assertEqual(tcp.dport, 56789)
1565             self.assertEqual(tcp.sport, 12345)
1566             self.assert_packet_checksums_valid(p)
1567         except:
1568             self.logger.error(ppp("Unexpected or invalid packet:", p))
1569             raise
1570
1571         sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
1572         self.assertEqual(len(sessions), 0)
1573         flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1574         self.vapi.nat44_ei_add_del_identity_mapping(
1575             ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
1576             flags=flags, vrf_id=1, is_add=1)
1577         identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
1578         self.assertEqual(len(identity_mappings), 2)
1579
1580     def test_multiple_inside_interfaces(self):
1581         """ NAT44EI multiple non-overlapping address space inside interfaces
1582         """
1583
1584         self.nat44_add_address(self.nat_addr)
1585         flags = self.config_flags.NAT44_EI_IF_INSIDE
1586         self.vapi.nat44_ei_interface_add_del_feature(
1587             sw_if_index=self.pg0.sw_if_index,
1588             flags=flags, is_add=1)
1589         self.vapi.nat44_ei_interface_add_del_feature(
1590             sw_if_index=self.pg1.sw_if_index,
1591             flags=flags, is_add=1)
1592         self.vapi.nat44_ei_interface_add_del_feature(
1593             sw_if_index=self.pg3.sw_if_index,
1594             is_add=1)
1595
1596         # between two NAT44EI inside interfaces (no translation)
1597         pkts = self.create_stream_in(self.pg0, self.pg1)
1598         self.pg0.add_stream(pkts)
1599         self.pg_enable_capture(self.pg_interfaces)
1600         self.pg_start()
1601         capture = self.pg1.get_capture(len(pkts))
1602         self.verify_capture_no_translation(capture, self.pg0, self.pg1)
1603
1604         # from inside to interface without translation
1605         pkts = self.create_stream_in(self.pg0, self.pg2)
1606         self.pg0.add_stream(pkts)
1607         self.pg_enable_capture(self.pg_interfaces)
1608         self.pg_start()
1609         capture = self.pg2.get_capture(len(pkts))
1610         self.verify_capture_no_translation(capture, self.pg0, self.pg2)
1611
1612         # in2out 1st interface
1613         pkts = self.create_stream_in(self.pg0, self.pg3)
1614         self.pg0.add_stream(pkts)
1615         self.pg_enable_capture(self.pg_interfaces)
1616         self.pg_start()
1617         capture = self.pg3.get_capture(len(pkts))
1618         self.verify_capture_out(capture)
1619
1620         # out2in 1st interface
1621         pkts = self.create_stream_out(self.pg3)
1622         self.pg3.add_stream(pkts)
1623         self.pg_enable_capture(self.pg_interfaces)
1624         self.pg_start()
1625         capture = self.pg0.get_capture(len(pkts))
1626         self.verify_capture_in(capture, self.pg0)
1627
1628         # in2out 2nd interface
1629         pkts = self.create_stream_in(self.pg1, self.pg3)
1630         self.pg1.add_stream(pkts)
1631         self.pg_enable_capture(self.pg_interfaces)
1632         self.pg_start()
1633         capture = self.pg3.get_capture(len(pkts))
1634         self.verify_capture_out(capture)
1635
1636         # out2in 2nd interface
1637         pkts = self.create_stream_out(self.pg3)
1638         self.pg3.add_stream(pkts)
1639         self.pg_enable_capture(self.pg_interfaces)
1640         self.pg_start()
1641         capture = self.pg1.get_capture(len(pkts))
1642         self.verify_capture_in(capture, self.pg1)
1643
1644     def test_inside_overlapping_interfaces(self):
1645         """ NAT44EI multiple inside interfaces with overlapping address space
1646         """
1647
1648         static_nat_ip = "10.0.0.10"
1649         self.nat44_add_address(self.nat_addr)
1650         flags = self.config_flags.NAT44_EI_IF_INSIDE
1651         self.vapi.nat44_ei_interface_add_del_feature(
1652             sw_if_index=self.pg3.sw_if_index,
1653             is_add=1)
1654         self.vapi.nat44_ei_interface_add_del_feature(
1655             sw_if_index=self.pg4.sw_if_index,
1656             flags=flags, is_add=1)
1657         self.vapi.nat44_ei_interface_add_del_feature(
1658             sw_if_index=self.pg5.sw_if_index,
1659             flags=flags, is_add=1)
1660         self.vapi.nat44_ei_interface_add_del_feature(
1661             sw_if_index=self.pg6.sw_if_index,
1662             flags=flags, is_add=1)
1663         self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
1664                                       vrf_id=20)
1665
1666         # between NAT44EI inside interfaces with same VRF (no translation)
1667         pkts = self.create_stream_in(self.pg4, self.pg5)
1668         self.pg4.add_stream(pkts)
1669         self.pg_enable_capture(self.pg_interfaces)
1670         self.pg_start()
1671         capture = self.pg5.get_capture(len(pkts))
1672         self.verify_capture_no_translation(capture, self.pg4, self.pg5)
1673
1674         # between NAT44EI inside interfaces with different VRF (hairpinning)
1675         p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
1676              IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
1677              TCP(sport=1234, dport=5678))
1678         self.pg4.add_stream(p)
1679         self.pg_enable_capture(self.pg_interfaces)
1680         self.pg_start()
1681         capture = self.pg6.get_capture(1)
1682         p = capture[0]
1683         try:
1684             ip = p[IP]
1685             tcp = p[TCP]
1686             self.assertEqual(ip.src, self.nat_addr)
1687             self.assertEqual(ip.dst, self.pg6.remote_ip4)
1688             self.assertNotEqual(tcp.sport, 1234)
1689             self.assertEqual(tcp.dport, 5678)
1690         except:
1691             self.logger.error(ppp("Unexpected or invalid packet:", p))
1692             raise
1693
1694         # in2out 1st interface
1695         pkts = self.create_stream_in(self.pg4, self.pg3)
1696         self.pg4.add_stream(pkts)
1697         self.pg_enable_capture(self.pg_interfaces)
1698         self.pg_start()
1699         capture = self.pg3.get_capture(len(pkts))
1700         self.verify_capture_out(capture)
1701
1702         # out2in 1st interface
1703         pkts = self.create_stream_out(self.pg3)
1704         self.pg3.add_stream(pkts)
1705         self.pg_enable_capture(self.pg_interfaces)
1706         self.pg_start()
1707         capture = self.pg4.get_capture(len(pkts))
1708         self.verify_capture_in(capture, self.pg4)
1709
1710         # in2out 2nd interface
1711         pkts = self.create_stream_in(self.pg5, self.pg3)
1712         self.pg5.add_stream(pkts)
1713         self.pg_enable_capture(self.pg_interfaces)
1714         self.pg_start()
1715         capture = self.pg3.get_capture(len(pkts))
1716         self.verify_capture_out(capture)
1717
1718         # out2in 2nd interface
1719         pkts = self.create_stream_out(self.pg3)
1720         self.pg3.add_stream(pkts)
1721         self.pg_enable_capture(self.pg_interfaces)
1722         self.pg_start()
1723         capture = self.pg5.get_capture(len(pkts))
1724         self.verify_capture_in(capture, self.pg5)
1725
1726         # pg5 session dump
1727         addresses = self.vapi.nat44_ei_address_dump()
1728         self.assertEqual(len(addresses), 1)
1729         sessions = self.vapi.nat44_ei_user_session_dump(
1730             self.pg5.remote_ip4, 10)
1731         self.assertEqual(len(sessions), 3)
1732         for session in sessions:
1733             self.assertFalse(session.flags &
1734                              self.config_flags.NAT44_EI_STATIC_MAPPING)
1735             self.assertEqual(str(session.inside_ip_address),
1736                              self.pg5.remote_ip4)
1737             self.assertEqual(session.outside_ip_address,
1738                              addresses[0].ip_address)
1739         self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
1740         self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
1741         self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
1742         self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
1743         self.assertEqual(sessions[1].inside_port, self.udp_port_in)
1744         self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
1745         self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
1746         self.assertEqual(sessions[1].outside_port, self.udp_port_out)
1747         self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
1748
1749         # in2out 3rd interface
1750         pkts = self.create_stream_in(self.pg6, self.pg3)
1751         self.pg6.add_stream(pkts)
1752         self.pg_enable_capture(self.pg_interfaces)
1753         self.pg_start()
1754         capture = self.pg3.get_capture(len(pkts))
1755         self.verify_capture_out(capture, static_nat_ip, True)
1756
1757         # out2in 3rd interface
1758         pkts = self.create_stream_out(self.pg3, static_nat_ip)
1759         self.pg3.add_stream(pkts)
1760         self.pg_enable_capture(self.pg_interfaces)
1761         self.pg_start()
1762         capture = self.pg6.get_capture(len(pkts))
1763         self.verify_capture_in(capture, self.pg6)
1764
1765         # general user and session dump verifications
1766         users = self.vapi.nat44_ei_user_dump()
1767         self.assertGreaterEqual(len(users), 3)
1768         addresses = self.vapi.nat44_ei_address_dump()
1769         self.assertEqual(len(addresses), 1)
1770         for user in users:
1771             sessions = self.vapi.nat44_ei_user_session_dump(user.ip_address,
1772                                                             user.vrf_id)
1773             for session in sessions:
1774                 self.assertEqual(user.ip_address, session.inside_ip_address)
1775                 self.assertTrue(session.total_bytes > session.total_pkts > 0)
1776                 self.assertTrue(session.protocol in
1777                                 [IP_PROTOS.tcp, IP_PROTOS.udp,
1778                                  IP_PROTOS.icmp])
1779
1780         # pg4 session dump
1781         sessions = self.vapi.nat44_ei_user_session_dump(
1782             self.pg4.remote_ip4, 10)
1783         self.assertGreaterEqual(len(sessions), 4)
1784         for session in sessions:
1785             self.assertFalse(
1786                 session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
1787             self.assertEqual(str(session.inside_ip_address),
1788                              self.pg4.remote_ip4)
1789             self.assertEqual(session.outside_ip_address,
1790                              addresses[0].ip_address)
1791
1792         # pg6 session dump
1793         sessions = self.vapi.nat44_ei_user_session_dump(
1794             self.pg6.remote_ip4, 20)
1795         self.assertGreaterEqual(len(sessions), 3)
1796         for session in sessions:
1797             self.assertTrue(
1798                 session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
1799             self.assertEqual(str(session.inside_ip_address),
1800                              self.pg6.remote_ip4)
1801             self.assertEqual(str(session.outside_ip_address),
1802                              static_nat_ip)
1803             self.assertTrue(session.inside_port in
1804                             [self.tcp_port_in, self.udp_port_in,
1805                              self.icmp_id_in])
1806
1807     def test_hairpinning(self):
1808         """ NAT44EI hairpinning - 1:1 NAPT """
1809
1810         host = self.pg0.remote_hosts[0]
1811         server = self.pg0.remote_hosts[1]
1812         host_in_port = 1234
1813         host_out_port = 0
1814         server_in_port = 5678
1815         server_out_port = 8765
1816
1817         self.nat44_add_address(self.nat_addr)
1818         flags = self.config_flags.NAT44_EI_IF_INSIDE
1819         self.vapi.nat44_ei_interface_add_del_feature(
1820             sw_if_index=self.pg0.sw_if_index,
1821             flags=flags, is_add=1)
1822         self.vapi.nat44_ei_interface_add_del_feature(
1823             sw_if_index=self.pg1.sw_if_index,
1824             is_add=1)
1825
1826         # add static mapping for server
1827         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
1828                                       server_in_port, server_out_port,
1829                                       proto=IP_PROTOS.tcp)
1830
1831         cnt = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
1832         # send packet from host to server
1833         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
1834              IP(src=host.ip4, dst=self.nat_addr) /
1835              TCP(sport=host_in_port, dport=server_out_port))
1836         self.pg0.add_stream(p)
1837         self.pg_enable_capture(self.pg_interfaces)
1838         self.pg_start()
1839         capture = self.pg0.get_capture(1)
1840         p = capture[0]
1841         try:
1842             ip = p[IP]
1843             tcp = p[TCP]
1844             self.assertEqual(ip.src, self.nat_addr)
1845             self.assertEqual(ip.dst, server.ip4)
1846             self.assertNotEqual(tcp.sport, host_in_port)
1847             self.assertEqual(tcp.dport, server_in_port)
1848             self.assert_packet_checksums_valid(p)
1849             host_out_port = tcp.sport
1850         except:
1851             self.logger.error(ppp("Unexpected or invalid packet:", p))
1852             raise
1853
1854         after = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
1855         if_idx = self.pg0.sw_if_index
1856         self.assertEqual(after[if_idx] - cnt[if_idx], 1)
1857
1858         # send reply from server to host
1859         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1860              IP(src=server.ip4, dst=self.nat_addr) /
1861              TCP(sport=server_in_port, dport=host_out_port))
1862         self.pg0.add_stream(p)
1863         self.pg_enable_capture(self.pg_interfaces)
1864         self.pg_start()
1865         capture = self.pg0.get_capture(1)
1866         p = capture[0]
1867         try:
1868             ip = p[IP]
1869             tcp = p[TCP]
1870             self.assertEqual(ip.src, self.nat_addr)
1871             self.assertEqual(ip.dst, host.ip4)
1872             self.assertEqual(tcp.sport, server_out_port)
1873             self.assertEqual(tcp.dport, host_in_port)
1874             self.assert_packet_checksums_valid(p)
1875         except:
1876             self.logger.error(ppp("Unexpected or invalid packet:", p))
1877             raise
1878
1879         after = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
1880         if_idx = self.pg0.sw_if_index
1881         self.assertEqual(after[if_idx] - cnt[if_idx], 2)
1882
1883     def test_hairpinning2(self):
1884         """ NAT44EI hairpinning - 1:1 NAT"""
1885
1886         server1_nat_ip = "10.0.0.10"
1887         server2_nat_ip = "10.0.0.11"
1888         host = self.pg0.remote_hosts[0]
1889         server1 = self.pg0.remote_hosts[1]
1890         server2 = self.pg0.remote_hosts[2]
1891         server_tcp_port = 22
1892         server_udp_port = 20
1893
1894         self.nat44_add_address(self.nat_addr)
1895         flags = self.config_flags.NAT44_EI_IF_INSIDE
1896         self.vapi.nat44_ei_interface_add_del_feature(
1897             sw_if_index=self.pg0.sw_if_index,
1898             flags=flags, is_add=1)
1899         self.vapi.nat44_ei_interface_add_del_feature(
1900             sw_if_index=self.pg1.sw_if_index,
1901             is_add=1)
1902
1903         # add static mapping for servers
1904         self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
1905         self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
1906
1907         # host to server1
1908         pkts = []
1909         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1910              IP(src=host.ip4, dst=server1_nat_ip) /
1911              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
1912         pkts.append(p)
1913         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1914              IP(src=host.ip4, dst=server1_nat_ip) /
1915              UDP(sport=self.udp_port_in, dport=server_udp_port))
1916         pkts.append(p)
1917         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1918              IP(src=host.ip4, dst=server1_nat_ip) /
1919              ICMP(id=self.icmp_id_in, type='echo-request'))
1920         pkts.append(p)
1921         self.pg0.add_stream(pkts)
1922         self.pg_enable_capture(self.pg_interfaces)
1923         self.pg_start()
1924         capture = self.pg0.get_capture(len(pkts))
1925         for packet in capture:
1926             try:
1927                 self.assertEqual(packet[IP].src, self.nat_addr)
1928                 self.assertEqual(packet[IP].dst, server1.ip4)
1929                 if packet.haslayer(TCP):
1930                     self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
1931                     self.assertEqual(packet[TCP].dport, server_tcp_port)
1932                     self.tcp_port_out = packet[TCP].sport
1933                     self.assert_packet_checksums_valid(packet)
1934                 elif packet.haslayer(UDP):
1935                     self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
1936                     self.assertEqual(packet[UDP].dport, server_udp_port)
1937                     self.udp_port_out = packet[UDP].sport
1938                 else:
1939                     self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
1940                     self.icmp_id_out = packet[ICMP].id
1941             except:
1942                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1943                 raise
1944
1945         # server1 to host
1946         pkts = []
1947         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1948              IP(src=server1.ip4, dst=self.nat_addr) /
1949              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
1950         pkts.append(p)
1951         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1952              IP(src=server1.ip4, dst=self.nat_addr) /
1953              UDP(sport=server_udp_port, dport=self.udp_port_out))
1954         pkts.append(p)
1955         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1956              IP(src=server1.ip4, dst=self.nat_addr) /
1957              ICMP(id=self.icmp_id_out, type='echo-reply'))
1958         pkts.append(p)
1959         self.pg0.add_stream(pkts)
1960         self.pg_enable_capture(self.pg_interfaces)
1961         self.pg_start()
1962         capture = self.pg0.get_capture(len(pkts))
1963         for packet in capture:
1964             try:
1965                 self.assertEqual(packet[IP].src, server1_nat_ip)
1966                 self.assertEqual(packet[IP].dst, host.ip4)
1967                 if packet.haslayer(TCP):
1968                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
1969                     self.assertEqual(packet[TCP].sport, server_tcp_port)
1970                     self.assert_packet_checksums_valid(packet)
1971                 elif packet.haslayer(UDP):
1972                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
1973                     self.assertEqual(packet[UDP].sport, server_udp_port)
1974                 else:
1975                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
1976             except:
1977                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1978                 raise
1979
1980         # server2 to server1
1981         pkts = []
1982         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1983              IP(src=server2.ip4, dst=server1_nat_ip) /
1984              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
1985         pkts.append(p)
1986         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1987              IP(src=server2.ip4, dst=server1_nat_ip) /
1988              UDP(sport=self.udp_port_in, dport=server_udp_port))
1989         pkts.append(p)
1990         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1991              IP(src=server2.ip4, dst=server1_nat_ip) /
1992              ICMP(id=self.icmp_id_in, type='echo-request'))
1993         pkts.append(p)
1994         self.pg0.add_stream(pkts)
1995         self.pg_enable_capture(self.pg_interfaces)
1996         self.pg_start()
1997         capture = self.pg0.get_capture(len(pkts))
1998         for packet in capture:
1999             try:
2000                 self.assertEqual(packet[IP].src, server2_nat_ip)
2001                 self.assertEqual(packet[IP].dst, server1.ip4)
2002                 if packet.haslayer(TCP):
2003                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
2004                     self.assertEqual(packet[TCP].dport, server_tcp_port)
2005                     self.tcp_port_out = packet[TCP].sport
2006                     self.assert_packet_checksums_valid(packet)
2007                 elif packet.haslayer(UDP):
2008                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
2009                     self.assertEqual(packet[UDP].dport, server_udp_port)
2010                     self.udp_port_out = packet[UDP].sport
2011                 else:
2012                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2013                     self.icmp_id_out = packet[ICMP].id
2014             except:
2015                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2016                 raise
2017
2018         # server1 to server2
2019         pkts = []
2020         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2021              IP(src=server1.ip4, dst=server2_nat_ip) /
2022              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
2023         pkts.append(p)
2024         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2025              IP(src=server1.ip4, dst=server2_nat_ip) /
2026              UDP(sport=server_udp_port, dport=self.udp_port_out))
2027         pkts.append(p)
2028         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2029              IP(src=server1.ip4, dst=server2_nat_ip) /
2030              ICMP(id=self.icmp_id_out, type='echo-reply'))
2031         pkts.append(p)
2032         self.pg0.add_stream(pkts)
2033         self.pg_enable_capture(self.pg_interfaces)
2034         self.pg_start()
2035         capture = self.pg0.get_capture(len(pkts))
2036         for packet in capture:
2037             try:
2038                 self.assertEqual(packet[IP].src, server1_nat_ip)
2039                 self.assertEqual(packet[IP].dst, server2.ip4)
2040                 if packet.haslayer(TCP):
2041                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
2042                     self.assertEqual(packet[TCP].sport, server_tcp_port)
2043                     self.assert_packet_checksums_valid(packet)
2044                 elif packet.haslayer(UDP):
2045                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
2046                     self.assertEqual(packet[UDP].sport, server_udp_port)
2047                 else:
2048                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2049             except:
2050                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2051                 raise
2052
2053     def test_hairpinning_avoid_inf_loop(self):
2054         """ NAT44EI hairpinning - 1:1 NAPT avoid infinite loop """
2055
2056         host = self.pg0.remote_hosts[0]
2057         server = self.pg0.remote_hosts[1]
2058         host_in_port = 1234
2059         host_out_port = 0
2060         server_in_port = 5678
2061         server_out_port = 8765
2062
2063         self.nat44_add_address(self.nat_addr)
2064         flags = self.config_flags.NAT44_EI_IF_INSIDE
2065         self.vapi.nat44_ei_interface_add_del_feature(
2066             sw_if_index=self.pg0.sw_if_index,
2067             flags=flags, is_add=1)
2068         self.vapi.nat44_ei_interface_add_del_feature(
2069             sw_if_index=self.pg1.sw_if_index,
2070             is_add=1)
2071
2072         # add static mapping for server
2073         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
2074                                       server_in_port, server_out_port,
2075                                       proto=IP_PROTOS.tcp)
2076
2077         # add another static mapping that maps pg0.local_ip4 address to itself
2078         self.nat44_add_static_mapping(self.pg0.local_ip4, self.pg0.local_ip4)
2079
2080         # send packet from host to VPP (the packet should get dropped)
2081         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2082              IP(src=host.ip4, dst=self.pg0.local_ip4) /
2083              TCP(sport=host_in_port, dport=server_out_port))
2084         self.pg0.add_stream(p)
2085         self.pg_enable_capture(self.pg_interfaces)
2086         self.pg_start()
2087         # Here VPP used to crash due to an infinite loop
2088
2089         cnt = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
2090         # send packet from host to server
2091         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2092              IP(src=host.ip4, dst=self.nat_addr) /
2093              TCP(sport=host_in_port, dport=server_out_port))
2094         self.pg0.add_stream(p)
2095         self.pg_enable_capture(self.pg_interfaces)
2096         self.pg_start()
2097         capture = self.pg0.get_capture(1)
2098         p = capture[0]
2099         try:
2100             ip = p[IP]
2101             tcp = p[TCP]
2102             self.assertEqual(ip.src, self.nat_addr)
2103             self.assertEqual(ip.dst, server.ip4)
2104             self.assertNotEqual(tcp.sport, host_in_port)
2105             self.assertEqual(tcp.dport, server_in_port)
2106             self.assert_packet_checksums_valid(p)
2107             host_out_port = tcp.sport
2108         except:
2109             self.logger.error(ppp("Unexpected or invalid packet:", p))
2110             raise
2111
2112         after = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
2113         if_idx = self.pg0.sw_if_index
2114         self.assertEqual(after[if_idx] - cnt[if_idx], 1)
2115
2116         # send reply from server to host
2117         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
2118              IP(src=server.ip4, dst=self.nat_addr) /
2119              TCP(sport=server_in_port, dport=host_out_port))
2120         self.pg0.add_stream(p)
2121         self.pg_enable_capture(self.pg_interfaces)
2122         self.pg_start()
2123         capture = self.pg0.get_capture(1)
2124         p = capture[0]
2125         try:
2126             ip = p[IP]
2127             tcp = p[TCP]
2128             self.assertEqual(ip.src, self.nat_addr)
2129             self.assertEqual(ip.dst, host.ip4)
2130             self.assertEqual(tcp.sport, server_out_port)
2131             self.assertEqual(tcp.dport, host_in_port)
2132             self.assert_packet_checksums_valid(p)
2133         except:
2134             self.logger.error(ppp("Unexpected or invalid packet:", p))
2135             raise
2136
2137         after = self.statistics.get_counter('/nat44-ei/hairpinning')[0]
2138         if_idx = self.pg0.sw_if_index
2139         self.assertEqual(after[if_idx] - cnt[if_idx], 2)
2140
2141     def test_interface_addr(self):
2142         """ NAT44EI acquire addresses from interface """
2143         self.vapi.nat44_ei_add_del_interface_addr(
2144             is_add=1,
2145             sw_if_index=self.pg7.sw_if_index)
2146
2147         # no address in NAT pool
2148         addresses = self.vapi.nat44_ei_address_dump()
2149         self.assertEqual(0, len(addresses))
2150
2151         # configure interface address and check NAT address pool
2152         self.pg7.config_ip4()
2153         addresses = self.vapi.nat44_ei_address_dump()
2154         self.assertEqual(1, len(addresses))
2155         self.assertEqual(str(addresses[0].ip_address), self.pg7.local_ip4)
2156
2157         # remove interface address and check NAT address pool
2158         self.pg7.unconfig_ip4()
2159         addresses = self.vapi.nat44_ei_address_dump()
2160         self.assertEqual(0, len(addresses))
2161
2162     def test_interface_addr_static_mapping(self):
2163         """ NAT44EI Static mapping with addresses from interface """
2164         tag = "testTAG"
2165
2166         self.vapi.nat44_ei_add_del_interface_addr(
2167             is_add=1,
2168             sw_if_index=self.pg7.sw_if_index)
2169         self.nat44_add_static_mapping(
2170             '1.2.3.4',
2171             external_sw_if_index=self.pg7.sw_if_index,
2172             tag=tag)
2173
2174         # static mappings with external interface
2175         static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2176         self.assertEqual(1, len(static_mappings))
2177         self.assertEqual(self.pg7.sw_if_index,
2178                          static_mappings[0].external_sw_if_index)
2179         self.assertEqual(static_mappings[0].tag, tag)
2180
2181         # configure interface address and check static mappings
2182         self.pg7.config_ip4()
2183         static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2184         self.assertEqual(2, len(static_mappings))
2185         resolved = False
2186         for sm in static_mappings:
2187             if sm.external_sw_if_index == 0xFFFFFFFF:
2188                 self.assertEqual(str(sm.external_ip_address),
2189                                  self.pg7.local_ip4)
2190                 self.assertEqual(sm.tag, tag)
2191                 resolved = True
2192         self.assertTrue(resolved)
2193
2194         # remove interface address and check static mappings
2195         self.pg7.unconfig_ip4()
2196         static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2197         self.assertEqual(1, len(static_mappings))
2198         self.assertEqual(self.pg7.sw_if_index,
2199                          static_mappings[0].external_sw_if_index)
2200         self.assertEqual(static_mappings[0].tag, tag)
2201
2202         # configure interface address again and check static mappings
2203         self.pg7.config_ip4()
2204         static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2205         self.assertEqual(2, len(static_mappings))
2206         resolved = False
2207         for sm in static_mappings:
2208             if sm.external_sw_if_index == 0xFFFFFFFF:
2209                 self.assertEqual(str(sm.external_ip_address),
2210                                  self.pg7.local_ip4)
2211                 self.assertEqual(sm.tag, tag)
2212                 resolved = True
2213         self.assertTrue(resolved)
2214
2215         # remove static mapping
2216         self.nat44_add_static_mapping(
2217             '1.2.3.4',
2218             external_sw_if_index=self.pg7.sw_if_index,
2219             tag=tag,
2220             is_add=0)
2221         static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2222         self.assertEqual(0, len(static_mappings))
2223
2224     def test_interface_addr_identity_nat(self):
2225         """ NAT44EI Identity NAT with addresses from interface """
2226
2227         port = 53053
2228         self.vapi.nat44_ei_add_del_interface_addr(
2229             is_add=1,
2230             sw_if_index=self.pg7.sw_if_index)
2231         self.vapi.nat44_ei_add_del_identity_mapping(
2232             ip_address=b'0',
2233             sw_if_index=self.pg7.sw_if_index,
2234             port=port,
2235             protocol=IP_PROTOS.tcp,
2236             is_add=1)
2237
2238         # identity mappings with external interface
2239         identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2240         self.assertEqual(1, len(identity_mappings))
2241         self.assertEqual(self.pg7.sw_if_index,
2242                          identity_mappings[0].sw_if_index)
2243
2244         # configure interface address and check identity mappings
2245         self.pg7.config_ip4()
2246         identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2247         resolved = False
2248         self.assertEqual(2, len(identity_mappings))
2249         for sm in identity_mappings:
2250             if sm.sw_if_index == 0xFFFFFFFF:
2251                 self.assertEqual(str(identity_mappings[0].ip_address),
2252                                  self.pg7.local_ip4)
2253                 self.assertEqual(port, identity_mappings[0].port)
2254                 self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol)
2255                 resolved = True
2256         self.assertTrue(resolved)
2257
2258         # remove interface address and check identity mappings
2259         self.pg7.unconfig_ip4()
2260         identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2261         self.assertEqual(1, len(identity_mappings))
2262         self.assertEqual(self.pg7.sw_if_index,
2263                          identity_mappings[0].sw_if_index)
2264
2265     def test_ipfix_nat44_sess(self):
2266         """ NAT44EI IPFIX logging NAT44EI session created/deleted """
2267         self.ipfix_domain_id = 10
2268         self.ipfix_src_port = 20202
2269         collector_port = 30303
2270         bind_layers(UDP, IPFIX, dport=30303)
2271         self.nat44_add_address(self.nat_addr)
2272         flags = self.config_flags.NAT44_EI_IF_INSIDE
2273         self.vapi.nat44_ei_interface_add_del_feature(
2274             sw_if_index=self.pg0.sw_if_index,
2275             flags=flags, is_add=1)
2276         self.vapi.nat44_ei_interface_add_del_feature(
2277             sw_if_index=self.pg1.sw_if_index,
2278             is_add=1)
2279         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2280                                      src_address=self.pg3.local_ip4,
2281                                      path_mtu=512,
2282                                      template_interval=10,
2283                                      collector_port=collector_port)
2284         self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2285                                                 src_port=self.ipfix_src_port,
2286                                                 enable=1)
2287
2288         pkts = self.create_stream_in(self.pg0, self.pg1)
2289         self.pg0.add_stream(pkts)
2290         self.pg_enable_capture(self.pg_interfaces)
2291         self.pg_start()
2292         capture = self.pg1.get_capture(len(pkts))
2293         self.verify_capture_out(capture)
2294         self.nat44_add_address(self.nat_addr, is_add=0)
2295         self.vapi.ipfix_flush()
2296         capture = self.pg3.get_capture(7)
2297         ipfix = IPFIXDecoder()
2298         # first load template
2299         for p in capture:
2300             self.assertTrue(p.haslayer(IPFIX))
2301             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2302             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2303             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2304             self.assertEqual(p[UDP].dport, collector_port)
2305             self.assertEqual(p[IPFIX].observationDomainID,
2306                              self.ipfix_domain_id)
2307             if p.haslayer(Template):
2308                 ipfix.add_template(p.getlayer(Template))
2309         # verify events in data set
2310         for p in capture:
2311             if p.haslayer(Data):
2312                 data = ipfix.decode_data_set(p.getlayer(Set))
2313                 self.verify_ipfix_nat44_ses(data)
2314
2315     def test_ipfix_addr_exhausted(self):
2316         """ NAT44EI IPFIX logging NAT addresses exhausted """
2317         flags = self.config_flags.NAT44_EI_IF_INSIDE
2318         self.vapi.nat44_ei_interface_add_del_feature(
2319             sw_if_index=self.pg0.sw_if_index,
2320             flags=flags, is_add=1)
2321         self.vapi.nat44_ei_interface_add_del_feature(
2322             sw_if_index=self.pg1.sw_if_index,
2323             is_add=1)
2324         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2325                                      src_address=self.pg3.local_ip4,
2326                                      path_mtu=512,
2327                                      template_interval=10)
2328         self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2329                                                 src_port=self.ipfix_src_port,
2330                                                 enable=1)
2331
2332         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2333              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2334              TCP(sport=3025))
2335         self.pg0.add_stream(p)
2336         self.pg_enable_capture(self.pg_interfaces)
2337         self.pg_start()
2338         self.pg1.assert_nothing_captured()
2339         sleep(1)
2340         self.vapi.ipfix_flush()
2341         capture = self.pg3.get_capture(7)
2342         ipfix = IPFIXDecoder()
2343         # first load template
2344         for p in capture:
2345             self.assertTrue(p.haslayer(IPFIX))
2346             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2347             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2348             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2349             self.assertEqual(p[UDP].dport, 4739)
2350             self.assertEqual(p[IPFIX].observationDomainID,
2351                              self.ipfix_domain_id)
2352             if p.haslayer(Template):
2353                 ipfix.add_template(p.getlayer(Template))
2354         # verify events in data set
2355         for p in capture:
2356             if p.haslayer(Data):
2357                 data = ipfix.decode_data_set(p.getlayer(Set))
2358                 self.verify_ipfix_addr_exhausted(data)
2359
2360     def test_ipfix_max_sessions(self):
2361         """ NAT44EI IPFIX logging maximum session entries exceeded """
2362         self.nat44_add_address(self.nat_addr)
2363         flags = self.config_flags.NAT44_EI_IF_INSIDE
2364         self.vapi.nat44_ei_interface_add_del_feature(
2365             sw_if_index=self.pg0.sw_if_index,
2366             flags=flags, is_add=1)
2367         self.vapi.nat44_ei_interface_add_del_feature(
2368             sw_if_index=self.pg1.sw_if_index,
2369             is_add=1)
2370
2371         max_sessions = self.max_translations
2372
2373         pkts = []
2374         for i in range(0, max_sessions):
2375             src = "10.10.%u.%u" % ((i & 0xFF00) >> 8, i & 0xFF)
2376             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2377                  IP(src=src, dst=self.pg1.remote_ip4) /
2378                  TCP(sport=1025))
2379             pkts.append(p)
2380         self.pg0.add_stream(pkts)
2381         self.pg_enable_capture(self.pg_interfaces)
2382         self.pg_start()
2383
2384         self.pg1.get_capture(max_sessions)
2385         self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2386                                      src_address=self.pg3.local_ip4,
2387                                      path_mtu=512,
2388                                      template_interval=10)
2389         self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2390                                                 src_port=self.ipfix_src_port,
2391                                                 enable=1)
2392
2393         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2394              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2395              TCP(sport=1025))
2396         self.pg0.add_stream(p)
2397         self.pg_enable_capture(self.pg_interfaces)
2398         self.pg_start()
2399         self.pg1.assert_nothing_captured()
2400         sleep(1)
2401         self.vapi.ipfix_flush()
2402         capture = self.pg3.get_capture(7)
2403         ipfix = IPFIXDecoder()
2404         # first load template
2405         for p in capture:
2406             self.assertTrue(p.haslayer(IPFIX))
2407             self.assertEqual(p[IP].src, self.pg3.local_ip4)
2408             self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2409             self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2410             self.assertEqual(p[UDP].dport, 4739)
2411             self.assertEqual(p[IPFIX].observationDomainID,
2412                              self.ipfix_domain_id)
2413             if p.haslayer(Template):
2414                 ipfix.add_template(p.getlayer(Template))
2415         # verify events in data set
2416         for p in capture:
2417             if p.haslayer(Data):
2418                 data = ipfix.decode_data_set(p.getlayer(Set))
2419                 self.verify_ipfix_max_sessions(data, max_sessions)
2420
2421     def test_syslog_apmap(self):
2422         """ NAT44EI syslog address and port mapping creation and deletion """
2423         self.vapi.syslog_set_filter(
2424             self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_INFO)
2425         self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
2426         self.nat44_add_address(self.nat_addr)
2427         flags = self.config_flags.NAT44_EI_IF_INSIDE
2428         self.vapi.nat44_ei_interface_add_del_feature(
2429             sw_if_index=self.pg0.sw_if_index,
2430             flags=flags, is_add=1)
2431         self.vapi.nat44_ei_interface_add_del_feature(
2432             sw_if_index=self.pg1.sw_if_index,
2433             is_add=1)
2434
2435         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2436              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2437              TCP(sport=self.tcp_port_in, dport=20))
2438         self.pg0.add_stream(p)
2439         self.pg_enable_capture(self.pg_interfaces)
2440         self.pg_start()
2441         capture = self.pg1.get_capture(1)
2442         self.tcp_port_out = capture[0][TCP].sport
2443         capture = self.pg3.get_capture(1)
2444         self.verify_syslog_apmap(capture[0][Raw].load)
2445
2446         self.pg_enable_capture(self.pg_interfaces)
2447         self.pg_start()
2448         self.nat44_add_address(self.nat_addr, is_add=0)
2449         capture = self.pg3.get_capture(1)
2450         self.verify_syslog_apmap(capture[0][Raw].load, False)
2451
2452     def test_pool_addr_fib(self):
2453         """ NAT44EI add pool addresses to FIB """
2454         static_addr = '10.0.0.10'
2455         self.nat44_add_address(self.nat_addr)
2456         flags = self.config_flags.NAT44_EI_IF_INSIDE
2457         self.vapi.nat44_ei_interface_add_del_feature(
2458             sw_if_index=self.pg0.sw_if_index,
2459             flags=flags, is_add=1)
2460         self.vapi.nat44_ei_interface_add_del_feature(
2461             sw_if_index=self.pg1.sw_if_index,
2462             is_add=1)
2463         self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
2464
2465         # NAT44EI address
2466         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2467              ARP(op=ARP.who_has, pdst=self.nat_addr,
2468                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2469         self.pg1.add_stream(p)
2470         self.pg_enable_capture(self.pg_interfaces)
2471         self.pg_start()
2472         capture = self.pg1.get_capture(1)
2473         self.assertTrue(capture[0].haslayer(ARP))
2474         self.assertTrue(capture[0][ARP].op, ARP.is_at)
2475
2476         # 1:1 NAT address
2477         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2478              ARP(op=ARP.who_has, pdst=static_addr,
2479                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2480         self.pg1.add_stream(p)
2481         self.pg_enable_capture(self.pg_interfaces)
2482         self.pg_start()
2483         capture = self.pg1.get_capture(1)
2484         self.assertTrue(capture[0].haslayer(ARP))
2485         self.assertTrue(capture[0][ARP].op, ARP.is_at)
2486
2487         # send ARP to non-NAT44EI interface
2488         p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2489              ARP(op=ARP.who_has, pdst=self.nat_addr,
2490                  psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
2491         self.pg2.add_stream(p)
2492         self.pg_enable_capture(self.pg_interfaces)
2493         self.pg_start()
2494         self.pg1.assert_nothing_captured()
2495
2496         # remove addresses and verify
2497         self.nat44_add_address(self.nat_addr, is_add=0)
2498         self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
2499                                       is_add=0)
2500
2501         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2502              ARP(op=ARP.who_has, pdst=self.nat_addr,
2503                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2504         self.pg1.add_stream(p)
2505         self.pg_enable_capture(self.pg_interfaces)
2506         self.pg_start()
2507         self.pg1.assert_nothing_captured()
2508
2509         p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2510              ARP(op=ARP.who_has, pdst=static_addr,
2511                  psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2512         self.pg1.add_stream(p)
2513         self.pg_enable_capture(self.pg_interfaces)
2514         self.pg_start()
2515         self.pg1.assert_nothing_captured()
2516
2517     def test_vrf_mode(self):
2518         """ NAT44EI tenant VRF aware address pool mode """
2519
2520         vrf_id1 = 1
2521         vrf_id2 = 2
2522         nat_ip1 = "10.0.0.10"
2523         nat_ip2 = "10.0.0.11"
2524
2525         self.pg0.unconfig_ip4()
2526         self.pg1.unconfig_ip4()
2527         self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
2528         self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
2529         self.pg0.set_table_ip4(vrf_id1)
2530         self.pg1.set_table_ip4(vrf_id2)
2531         self.pg0.config_ip4()
2532         self.pg1.config_ip4()
2533         self.pg0.resolve_arp()
2534         self.pg1.resolve_arp()
2535
2536         self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
2537         self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
2538         flags = self.config_flags.NAT44_EI_IF_INSIDE
2539         self.vapi.nat44_ei_interface_add_del_feature(
2540             sw_if_index=self.pg0.sw_if_index,
2541             flags=flags, is_add=1)
2542         self.vapi.nat44_ei_interface_add_del_feature(
2543             sw_if_index=self.pg1.sw_if_index,
2544             flags=flags, is_add=1)
2545         self.vapi.nat44_ei_interface_add_del_feature(
2546             sw_if_index=self.pg2.sw_if_index,
2547             is_add=1)
2548
2549         try:
2550             # first VRF
2551             pkts = self.create_stream_in(self.pg0, self.pg2)
2552             self.pg0.add_stream(pkts)
2553             self.pg_enable_capture(self.pg_interfaces)
2554             self.pg_start()
2555             capture = self.pg2.get_capture(len(pkts))
2556             self.verify_capture_out(capture, nat_ip1)
2557
2558             # second VRF
2559             pkts = self.create_stream_in(self.pg1, self.pg2)
2560             self.pg1.add_stream(pkts)
2561             self.pg_enable_capture(self.pg_interfaces)
2562             self.pg_start()
2563             capture = self.pg2.get_capture(len(pkts))
2564             self.verify_capture_out(capture, nat_ip2)
2565
2566         finally:
2567             self.pg0.unconfig_ip4()
2568             self.pg1.unconfig_ip4()
2569             self.pg0.set_table_ip4(0)
2570             self.pg1.set_table_ip4(0)
2571             self.pg0.config_ip4()
2572             self.pg1.config_ip4()
2573             self.pg0.resolve_arp()
2574             self.pg1.resolve_arp()
2575             self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id1})
2576             self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id2})
2577
2578     def test_vrf_feature_independent(self):
2579         """ NAT44EI tenant VRF independent address pool mode """
2580
2581         nat_ip1 = "10.0.0.10"
2582         nat_ip2 = "10.0.0.11"
2583
2584         self.nat44_add_address(nat_ip1)
2585         self.nat44_add_address(nat_ip2, vrf_id=99)
2586         flags = self.config_flags.NAT44_EI_IF_INSIDE
2587         self.vapi.nat44_ei_interface_add_del_feature(
2588             sw_if_index=self.pg0.sw_if_index,
2589             flags=flags, is_add=1)
2590         self.vapi.nat44_ei_interface_add_del_feature(
2591             sw_if_index=self.pg1.sw_if_index,
2592             flags=flags, is_add=1)
2593         self.vapi.nat44_ei_interface_add_del_feature(
2594             sw_if_index=self.pg2.sw_if_index,
2595             is_add=1)
2596
2597         # first VRF
2598         pkts = self.create_stream_in(self.pg0, self.pg2)
2599         self.pg0.add_stream(pkts)
2600         self.pg_enable_capture(self.pg_interfaces)
2601         self.pg_start()
2602         capture = self.pg2.get_capture(len(pkts))
2603         self.verify_capture_out(capture, nat_ip1)
2604
2605         # second VRF
2606         pkts = self.create_stream_in(self.pg1, self.pg2)
2607         self.pg1.add_stream(pkts)
2608         self.pg_enable_capture(self.pg_interfaces)
2609         self.pg_start()
2610         capture = self.pg2.get_capture(len(pkts))
2611         self.verify_capture_out(capture, nat_ip1)
2612
2613     def test_dynamic_ipless_interfaces(self):
2614         """ NAT44EI interfaces without configured IP address """
2615         self.create_routes_and_neigbors()
2616         self.nat44_add_address(self.nat_addr)
2617         flags = self.config_flags.NAT44_EI_IF_INSIDE
2618         self.vapi.nat44_ei_interface_add_del_feature(
2619             sw_if_index=self.pg7.sw_if_index,
2620             flags=flags, is_add=1)
2621         self.vapi.nat44_ei_interface_add_del_feature(
2622             sw_if_index=self.pg8.sw_if_index,
2623             is_add=1)
2624
2625         # in2out
2626         pkts = self.create_stream_in(self.pg7, self.pg8)
2627         self.pg7.add_stream(pkts)
2628         self.pg_enable_capture(self.pg_interfaces)
2629         self.pg_start()
2630         capture = self.pg8.get_capture(len(pkts))
2631         self.verify_capture_out(capture)
2632
2633         # out2in
2634         pkts = self.create_stream_out(self.pg8, self.nat_addr)
2635         self.pg8.add_stream(pkts)
2636         self.pg_enable_capture(self.pg_interfaces)
2637         self.pg_start()
2638         capture = self.pg7.get_capture(len(pkts))
2639         self.verify_capture_in(capture, self.pg7)
2640
2641     def test_static_ipless_interfaces(self):
2642         """ NAT44EI interfaces without configured IP address - 1:1 NAT """
2643
2644         self.create_routes_and_neigbors()
2645         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
2646         flags = self.config_flags.NAT44_EI_IF_INSIDE
2647         self.vapi.nat44_ei_interface_add_del_feature(
2648             sw_if_index=self.pg7.sw_if_index,
2649             flags=flags, is_add=1)
2650         self.vapi.nat44_ei_interface_add_del_feature(
2651             sw_if_index=self.pg8.sw_if_index,
2652             is_add=1)
2653
2654         # out2in
2655         pkts = self.create_stream_out(self.pg8)
2656         self.pg8.add_stream(pkts)
2657         self.pg_enable_capture(self.pg_interfaces)
2658         self.pg_start()
2659         capture = self.pg7.get_capture(len(pkts))
2660         self.verify_capture_in(capture, self.pg7)
2661
2662         # in2out
2663         pkts = self.create_stream_in(self.pg7, self.pg8)
2664         self.pg7.add_stream(pkts)
2665         self.pg_enable_capture(self.pg_interfaces)
2666         self.pg_start()
2667         capture = self.pg8.get_capture(len(pkts))
2668         self.verify_capture_out(capture, self.nat_addr, True)
2669
2670     def test_static_with_port_ipless_interfaces(self):
2671         """ NAT44EI interfaces without configured IP address - 1:1 NAPT """
2672
2673         self.tcp_port_out = 30606
2674         self.udp_port_out = 30607
2675         self.icmp_id_out = 30608
2676
2677         self.create_routes_and_neigbors()
2678         self.nat44_add_address(self.nat_addr)
2679         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2680                                       self.tcp_port_in, self.tcp_port_out,
2681                                       proto=IP_PROTOS.tcp)
2682         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2683                                       self.udp_port_in, self.udp_port_out,
2684                                       proto=IP_PROTOS.udp)
2685         self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2686                                       self.icmp_id_in, self.icmp_id_out,
2687                                       proto=IP_PROTOS.icmp)
2688         flags = self.config_flags.NAT44_EI_IF_INSIDE
2689         self.vapi.nat44_ei_interface_add_del_feature(
2690             sw_if_index=self.pg7.sw_if_index,
2691             flags=flags, is_add=1)
2692         self.vapi.nat44_ei_interface_add_del_feature(
2693             sw_if_index=self.pg8.sw_if_index,
2694             is_add=1)
2695
2696         # out2in
2697         pkts = self.create_stream_out(self.pg8)
2698         self.pg8.add_stream(pkts)
2699         self.pg_enable_capture(self.pg_interfaces)
2700         self.pg_start()
2701         capture = self.pg7.get_capture(len(pkts))
2702         self.verify_capture_in(capture, self.pg7)
2703
2704         # in2out
2705         pkts = self.create_stream_in(self.pg7, self.pg8)
2706         self.pg7.add_stream(pkts)
2707         self.pg_enable_capture(self.pg_interfaces)
2708         self.pg_start()
2709         capture = self.pg8.get_capture(len(pkts))
2710         self.verify_capture_out(capture)
2711
2712     def test_static_unknown_proto(self):
2713         """ NAT44EI 1:1 translate packet with unknown protocol """
2714         nat_ip = "10.0.0.10"
2715         self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
2716         flags = self.config_flags.NAT44_EI_IF_INSIDE
2717         self.vapi.nat44_ei_interface_add_del_feature(
2718             sw_if_index=self.pg0.sw_if_index,
2719             flags=flags, is_add=1)
2720         self.vapi.nat44_ei_interface_add_del_feature(
2721             sw_if_index=self.pg1.sw_if_index,
2722             is_add=1)
2723
2724         # in2out
2725         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2726              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2727              GRE() /
2728              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2729              TCP(sport=1234, dport=1234))
2730         self.pg0.add_stream(p)
2731         self.pg_enable_capture(self.pg_interfaces)
2732         self.pg_start()
2733         p = self.pg1.get_capture(1)
2734         packet = p[0]
2735         try:
2736             self.assertEqual(packet[IP].src, nat_ip)
2737             self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
2738             self.assertEqual(packet.haslayer(GRE), 1)
2739             self.assert_packet_checksums_valid(packet)
2740         except:
2741             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2742             raise
2743
2744         # out2in
2745         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
2746              IP(src=self.pg1.remote_ip4, dst=nat_ip) /
2747              GRE() /
2748              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2749              TCP(sport=1234, dport=1234))
2750         self.pg1.add_stream(p)
2751         self.pg_enable_capture(self.pg_interfaces)
2752         self.pg_start()
2753         p = self.pg0.get_capture(1)
2754         packet = p[0]
2755         try:
2756             self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
2757             self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
2758             self.assertEqual(packet.haslayer(GRE), 1)
2759             self.assert_packet_checksums_valid(packet)
2760         except:
2761             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2762             raise
2763
2764     def test_hairpinning_static_unknown_proto(self):
2765         """ NAT44EI 1:1 translate packet with unknown protocol - hairpinning
2766         """
2767
2768         host = self.pg0.remote_hosts[0]
2769         server = self.pg0.remote_hosts[1]
2770
2771         host_nat_ip = "10.0.0.10"
2772         server_nat_ip = "10.0.0.11"
2773
2774         self.nat44_add_static_mapping(host.ip4, host_nat_ip)
2775         self.nat44_add_static_mapping(server.ip4, server_nat_ip)
2776         flags = self.config_flags.NAT44_EI_IF_INSIDE
2777         self.vapi.nat44_ei_interface_add_del_feature(
2778             sw_if_index=self.pg0.sw_if_index,
2779             flags=flags, is_add=1)
2780         self.vapi.nat44_ei_interface_add_del_feature(
2781             sw_if_index=self.pg1.sw_if_index,
2782             is_add=1)
2783
2784         # host to server
2785         p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
2786              IP(src=host.ip4, dst=server_nat_ip) /
2787              GRE() /
2788              IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2789              TCP(sport=1234, dport=1234))
2790         self.pg0.add_stream(p)
2791         self.pg_enable_capture(self.pg_interfaces)
2792         self.pg_start()
2793         p = self.pg0.get_capture(1)
2794         packet = p[0]
2795         try:
2796             self.assertEqual(packet[IP].src, host_nat_ip)
2797             self.assertEqual(packet[IP].dst, server.ip4)
2798             self.assertEqual(packet.haslayer(GRE), 1)
2799             self.assert_packet_checksums_valid(packet)
2800         except:
2801             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2802             raise
2803
2804         # server to host
2805         p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
2806              IP(src=server.ip4, dst=host_nat_ip) /
2807              GRE() /
2808              IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2809              TCP(sport=1234, dport=1234))
2810         self.pg0.add_stream(p)
2811         self.pg_enable_capture(self.pg_interfaces)
2812         self.pg_start()
2813         p = self.pg0.get_capture(1)
2814         packet = p[0]
2815         try:
2816             self.assertEqual(packet[IP].src, server_nat_ip)
2817             self.assertEqual(packet[IP].dst, host.ip4)
2818             self.assertEqual(packet.haslayer(GRE), 1)
2819             self.assert_packet_checksums_valid(packet)
2820         except:
2821             self.logger.error(ppp("Unexpected or invalid packet:", packet))
2822             raise
2823
2824     def test_output_feature(self):
2825         """ NAT44EI output feature (in2out postrouting) """
2826         self.nat44_add_address(self.nat_addr)
2827         flags = self.config_flags.NAT44_EI_IF_INSIDE
2828         self.vapi.nat44_ei_interface_add_del_output_feature(
2829             is_add=1, flags=flags,
2830             sw_if_index=self.pg0.sw_if_index)
2831         self.vapi.nat44_ei_interface_add_del_output_feature(
2832             is_add=1, flags=flags,
2833             sw_if_index=self.pg1.sw_if_index)
2834         self.vapi.nat44_ei_interface_add_del_output_feature(
2835             is_add=1,
2836             sw_if_index=self.pg3.sw_if_index)
2837
2838         # in2out
2839         pkts = self.create_stream_in(self.pg0, self.pg3)
2840         self.pg0.add_stream(pkts)
2841         self.pg_enable_capture(self.pg_interfaces)
2842         self.pg_start()
2843         capture = self.pg3.get_capture(len(pkts))
2844         self.verify_capture_out(capture)
2845
2846         # out2in
2847         pkts = self.create_stream_out(self.pg3)
2848         self.pg3.add_stream(pkts)
2849         self.pg_enable_capture(self.pg_interfaces)
2850         self.pg_start()
2851         capture = self.pg0.get_capture(len(pkts))
2852         self.verify_capture_in(capture, self.pg0)
2853
2854         # from non-NAT interface to NAT inside interface
2855         pkts = self.create_stream_in(self.pg2, self.pg0)
2856         self.pg2.add_stream(pkts)
2857         self.pg_enable_capture(self.pg_interfaces)
2858         self.pg_start()
2859         capture = self.pg0.get_capture(len(pkts))
2860         self.verify_capture_no_translation(capture, self.pg2, self.pg0)
2861
2862     def test_output_feature_vrf_aware(self):
2863         """ NAT44EI output feature VRF aware (in2out postrouting) """
2864         nat_ip_vrf10 = "10.0.0.10"
2865         nat_ip_vrf20 = "10.0.0.20"
2866
2867         r1 = VppIpRoute(self, self.pg3.remote_ip4, 32,
2868                         [VppRoutePath(self.pg3.remote_ip4,
2869                                       self.pg3.sw_if_index)],
2870                         table_id=10)
2871         r2 = VppIpRoute(self, self.pg3.remote_ip4, 32,
2872                         [VppRoutePath(self.pg3.remote_ip4,
2873                                       self.pg3.sw_if_index)],
2874                         table_id=20)
2875         r1.add_vpp_config()
2876         r2.add_vpp_config()
2877
2878         self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
2879         self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
2880         flags = self.config_flags.NAT44_EI_IF_INSIDE
2881         self.vapi.nat44_ei_interface_add_del_output_feature(
2882             is_add=1, flags=flags,
2883             sw_if_index=self.pg4.sw_if_index)
2884         self.vapi.nat44_ei_interface_add_del_output_feature(
2885             is_add=1, flags=flags,
2886             sw_if_index=self.pg6.sw_if_index)
2887         self.vapi.nat44_ei_interface_add_del_output_feature(
2888             is_add=1,
2889             sw_if_index=self.pg3.sw_if_index)
2890
2891         # in2out VRF 10
2892         pkts = self.create_stream_in(self.pg4, self.pg3)
2893         self.pg4.add_stream(pkts)
2894         self.pg_enable_capture(self.pg_interfaces)
2895         self.pg_start()
2896         capture = self.pg3.get_capture(len(pkts))
2897         self.verify_capture_out(capture, nat_ip=nat_ip_vrf10)
2898
2899         # out2in VRF 10
2900         pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10)
2901         self.pg3.add_stream(pkts)
2902         self.pg_enable_capture(self.pg_interfaces)
2903         self.pg_start()
2904         capture = self.pg4.get_capture(len(pkts))
2905         self.verify_capture_in(capture, self.pg4)
2906
2907         # in2out VRF 20
2908         pkts = self.create_stream_in(self.pg6, self.pg3)
2909         self.pg6.add_stream(pkts)
2910         self.pg_enable_capture(self.pg_interfaces)
2911         self.pg_start()
2912         capture = self.pg3.get_capture(len(pkts))
2913         self.verify_capture_out(capture, nat_ip=nat_ip_vrf20)
2914
2915         # out2in VRF 20
2916         pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20)
2917         self.pg3.add_stream(pkts)
2918         self.pg_enable_capture(self.pg_interfaces)
2919         self.pg_start()
2920         capture = self.pg6.get_capture(len(pkts))
2921         self.verify_capture_in(capture, self.pg6)
2922
2923     def test_output_feature_hairpinning(self):
2924         """ NAT44EI output feature hairpinning (in2out postrouting) """
2925         host = self.pg0.remote_hosts[0]
2926         server = self.pg0.remote_hosts[1]
2927         host_in_port = 1234
2928         host_out_port = 0
2929         server_in_port = 5678
2930         server_out_port = 8765
2931
2932         self.nat44_add_address(self.nat_addr)
2933         flags = self.config_flags.NAT44_EI_IF_INSIDE
2934         self.vapi.nat44_ei_interface_add_del_output_feature(
2935             is_add=1, flags=flags,
2936             sw_if_index=self.pg0.sw_if_index)
2937         self.vapi.nat44_ei_interface_add_del_output_feature(
2938             is_add=1,
2939             sw_if_index=self.pg1.sw_if_index)
2940
2941         # add static mapping for server
2942         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
2943                                       server_in_port, server_out_port,
2944                                       proto=IP_PROTOS.tcp)
2945
2946         # send packet from host to server
2947         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2948              IP(src=host.ip4, dst=self.nat_addr) /
2949              TCP(sport=host_in_port, dport=server_out_port))
2950         self.pg0.add_stream(p)
2951         self.pg_enable_capture(self.pg_interfaces)
2952         self.pg_start()
2953         capture = self.pg0.get_capture(1)
2954         p = capture[0]
2955         try:
2956             ip = p[IP]
2957             tcp = p[TCP]
2958             self.assertEqual(ip.src, self.nat_addr)
2959             self.assertEqual(ip.dst, server.ip4)
2960             self.assertNotEqual(tcp.sport, host_in_port)
2961             self.assertEqual(tcp.dport, server_in_port)
2962             self.assert_packet_checksums_valid(p)
2963             host_out_port = tcp.sport
2964         except:
2965             self.logger.error(ppp("Unexpected or invalid packet:", p))
2966             raise
2967
2968         # send reply from server to host
2969         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
2970              IP(src=server.ip4, dst=self.nat_addr) /
2971              TCP(sport=server_in_port, dport=host_out_port))
2972         self.pg0.add_stream(p)
2973         self.pg_enable_capture(self.pg_interfaces)
2974         self.pg_start()
2975         capture = self.pg0.get_capture(1)
2976         p = capture[0]
2977         try:
2978             ip = p[IP]
2979             tcp = p[TCP]
2980             self.assertEqual(ip.src, self.nat_addr)
2981             self.assertEqual(ip.dst, host.ip4)
2982             self.assertEqual(tcp.sport, server_out_port)
2983             self.assertEqual(tcp.dport, host_in_port)
2984             self.assert_packet_checksums_valid(p)
2985         except:
2986             self.logger.error(ppp("Unexpected or invalid packet:", p))
2987             raise
2988
2989     def test_one_armed_nat44(self):
2990         """ NAT44EI One armed NAT """
2991         remote_host = self.pg9.remote_hosts[0]
2992         local_host = self.pg9.remote_hosts[1]
2993         external_port = 0
2994
2995         self.nat44_add_address(self.nat_addr)
2996         flags = self.config_flags.NAT44_EI_IF_INSIDE
2997         self.vapi.nat44_ei_interface_add_del_feature(
2998             sw_if_index=self.pg9.sw_if_index,
2999             is_add=1)
3000         self.vapi.nat44_ei_interface_add_del_feature(
3001             sw_if_index=self.pg9.sw_if_index,
3002             flags=flags, is_add=1)
3003
3004         # in2out
3005         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3006              IP(src=local_host.ip4, dst=remote_host.ip4) /
3007              TCP(sport=12345, dport=80))
3008         self.pg9.add_stream(p)
3009         self.pg_enable_capture(self.pg_interfaces)
3010         self.pg_start()
3011         capture = self.pg9.get_capture(1)
3012         p = capture[0]
3013         try:
3014             ip = p[IP]
3015             tcp = p[TCP]
3016             self.assertEqual(ip.src, self.nat_addr)
3017             self.assertEqual(ip.dst, remote_host.ip4)
3018             self.assertNotEqual(tcp.sport, 12345)
3019             external_port = tcp.sport
3020             self.assertEqual(tcp.dport, 80)
3021             self.assert_packet_checksums_valid(p)
3022         except:
3023             self.logger.error(ppp("Unexpected or invalid packet:", p))
3024             raise
3025
3026         # out2in
3027         p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3028              IP(src=remote_host.ip4, dst=self.nat_addr) /
3029              TCP(sport=80, dport=external_port))
3030         self.pg9.add_stream(p)
3031         self.pg_enable_capture(self.pg_interfaces)
3032         self.pg_start()
3033         capture = self.pg9.get_capture(1)
3034         p = capture[0]
3035         try:
3036             ip = p[IP]
3037             tcp = p[TCP]
3038             self.assertEqual(ip.src, remote_host.ip4)
3039             self.assertEqual(ip.dst, local_host.ip4)
3040             self.assertEqual(tcp.sport, 80)
3041             self.assertEqual(tcp.dport, 12345)
3042             self.assert_packet_checksums_valid(p)
3043         except:
3044             self.logger.error(ppp("Unexpected or invalid packet:", p))
3045             raise
3046
3047         err = self.statistics.get_err_counter(
3048             '/err/nat44-ei-classify/next in2out')
3049         self.assertEqual(err, 1)
3050         err = self.statistics.get_err_counter(
3051             '/err/nat44-ei-classify/next out2in')
3052         self.assertEqual(err, 1)
3053
3054     def test_del_session(self):
3055         """ NAT44EI delete session """
3056         self.nat44_add_address(self.nat_addr)
3057         flags = self.config_flags.NAT44_EI_IF_INSIDE
3058         self.vapi.nat44_ei_interface_add_del_feature(
3059             sw_if_index=self.pg0.sw_if_index,
3060             flags=flags, is_add=1)
3061         self.vapi.nat44_ei_interface_add_del_feature(
3062             sw_if_index=self.pg1.sw_if_index,
3063             is_add=1)
3064
3065         pkts = self.create_stream_in(self.pg0, self.pg1)
3066         self.pg0.add_stream(pkts)
3067         self.pg_enable_capture(self.pg_interfaces)
3068         self.pg_start()
3069         self.pg1.get_capture(len(pkts))
3070
3071         sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
3072         nsessions = len(sessions)
3073
3074         self.vapi.nat44_ei_del_session(
3075             address=sessions[0].inside_ip_address,
3076             port=sessions[0].inside_port,
3077             protocol=sessions[0].protocol,
3078             flags=self.config_flags.NAT44_EI_IF_INSIDE)
3079
3080         self.vapi.nat44_ei_del_session(
3081             address=sessions[1].outside_ip_address,
3082             port=sessions[1].outside_port,
3083             protocol=sessions[1].protocol)
3084
3085         sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
3086         self.assertEqual(nsessions - len(sessions), 2)
3087
3088         self.vapi.nat44_ei_del_session(
3089             address=sessions[0].inside_ip_address,
3090             port=sessions[0].inside_port,
3091             protocol=sessions[0].protocol,
3092             flags=self.config_flags.NAT44_EI_IF_INSIDE)
3093
3094         self.verify_no_nat44_user()
3095
3096     def test_frag_in_order(self):
3097         """ NAT44EI translate fragments arriving in order """
3098
3099         self.nat44_add_address(self.nat_addr)
3100         flags = self.config_flags.NAT44_EI_IF_INSIDE
3101         self.vapi.nat44_ei_interface_add_del_feature(
3102             sw_if_index=self.pg0.sw_if_index,
3103             flags=flags, is_add=1)
3104         self.vapi.nat44_ei_interface_add_del_feature(
3105             sw_if_index=self.pg1.sw_if_index,
3106             is_add=1)
3107
3108         self.frag_in_order(proto=IP_PROTOS.tcp)
3109         self.frag_in_order(proto=IP_PROTOS.udp)
3110         self.frag_in_order(proto=IP_PROTOS.icmp)
3111
3112     def test_frag_forwarding(self):
3113         """ NAT44EI forwarding fragment test """
3114         self.vapi.nat44_ei_add_del_interface_addr(
3115             is_add=1,
3116             sw_if_index=self.pg1.sw_if_index)
3117         flags = self.config_flags.NAT44_EI_IF_INSIDE
3118         self.vapi.nat44_ei_interface_add_del_feature(
3119             sw_if_index=self.pg0.sw_if_index,
3120             flags=flags, is_add=1)
3121         self.vapi.nat44_ei_interface_add_del_feature(
3122             sw_if_index=self.pg1.sw_if_index,
3123             is_add=1)
3124         self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
3125
3126         data = b"A" * 16 + b"B" * 16 + b"C" * 3
3127         pkts = self.create_stream_frag(self.pg1,
3128                                        self.pg0.remote_ip4,
3129                                        4789,
3130                                        4789,
3131                                        data,
3132                                        proto=IP_PROTOS.udp)
3133         self.pg1.add_stream(pkts)
3134         self.pg_enable_capture(self.pg_interfaces)
3135         self.pg_start()
3136         frags = self.pg0.get_capture(len(pkts))
3137         p = self.reass_frags_and_verify(frags,
3138                                         self.pg1.remote_ip4,
3139                                         self.pg0.remote_ip4)
3140         self.assertEqual(p[UDP].sport, 4789)
3141         self.assertEqual(p[UDP].dport, 4789)
3142         self.assertEqual(data, p[Raw].load)
3143
3144     def test_reass_hairpinning(self):
3145         """ NAT44EI fragments hairpinning """
3146
3147         server_addr = self.pg0.remote_hosts[1].ip4
3148         host_in_port = random.randint(1025, 65535)
3149         server_in_port = random.randint(1025, 65535)
3150         server_out_port = random.randint(1025, 65535)
3151
3152         self.nat44_add_address(self.nat_addr)
3153         flags = self.config_flags.NAT44_EI_IF_INSIDE
3154         self.vapi.nat44_ei_interface_add_del_feature(
3155             sw_if_index=self.pg0.sw_if_index,
3156             flags=flags, is_add=1)
3157         self.vapi.nat44_ei_interface_add_del_feature(
3158             sw_if_index=self.pg1.sw_if_index,
3159             is_add=1)
3160         # add static mapping for server
3161         self.nat44_add_static_mapping(server_addr, self.nat_addr,
3162                                       server_in_port,
3163                                       server_out_port,
3164                                       proto=IP_PROTOS.tcp)
3165         self.nat44_add_static_mapping(server_addr, self.nat_addr,
3166                                       server_in_port,
3167                                       server_out_port,
3168                                       proto=IP_PROTOS.udp)
3169         self.nat44_add_static_mapping(server_addr, self.nat_addr)
3170
3171         self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3172                                host_in_port, proto=IP_PROTOS.tcp)
3173         self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3174                                host_in_port, proto=IP_PROTOS.udp)
3175         self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3176                                host_in_port, proto=IP_PROTOS.icmp)
3177
3178     def test_frag_out_of_order(self):
3179         """ NAT44EI translate fragments arriving out of order """
3180
3181         self.nat44_add_address(self.nat_addr)
3182         flags = self.config_flags.NAT44_EI_IF_INSIDE
3183         self.vapi.nat44_ei_interface_add_del_feature(
3184             sw_if_index=self.pg0.sw_if_index,
3185             flags=flags, is_add=1)
3186         self.vapi.nat44_ei_interface_add_del_feature(
3187             sw_if_index=self.pg1.sw_if_index,
3188             is_add=1)
3189
3190         self.frag_out_of_order(proto=IP_PROTOS.tcp)
3191         self.frag_out_of_order(proto=IP_PROTOS.udp)
3192         self.frag_out_of_order(proto=IP_PROTOS.icmp)
3193
3194     def test_port_restricted(self):
3195         """ NAT44EI Port restricted NAT44EI (MAP-E CE) """
3196         self.nat44_add_address(self.nat_addr)
3197         flags = self.config_flags.NAT44_EI_IF_INSIDE
3198         self.vapi.nat44_ei_interface_add_del_feature(
3199             sw_if_index=self.pg0.sw_if_index,
3200             flags=flags, is_add=1)
3201         self.vapi.nat44_ei_interface_add_del_feature(
3202             sw_if_index=self.pg1.sw_if_index,
3203             is_add=1)
3204         self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=1,
3205                                                        psid_offset=6,
3206                                                        psid_length=6,
3207                                                        psid=10)
3208
3209         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3210              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3211              TCP(sport=4567, dport=22))
3212         self.pg0.add_stream(p)
3213         self.pg_enable_capture(self.pg_interfaces)
3214         self.pg_start()
3215         capture = self.pg1.get_capture(1)
3216         p = capture[0]
3217         try:
3218             ip = p[IP]
3219             tcp = p[TCP]
3220             self.assertEqual(ip.dst, self.pg1.remote_ip4)
3221             self.assertEqual(ip.src, self.nat_addr)
3222             self.assertEqual(tcp.dport, 22)
3223             self.assertNotEqual(tcp.sport, 4567)
3224             self.assertEqual((tcp.sport >> 6) & 63, 10)
3225             self.assert_packet_checksums_valid(p)
3226         except:
3227             self.logger.error(ppp("Unexpected or invalid packet:", p))
3228             raise
3229
3230     def test_port_range(self):
3231         """ NAT44EI External address port range """
3232         self.nat44_add_address(self.nat_addr)
3233         flags = self.config_flags.NAT44_EI_IF_INSIDE
3234         self.vapi.nat44_ei_interface_add_del_feature(
3235             sw_if_index=self.pg0.sw_if_index,
3236             flags=flags, is_add=1)
3237         self.vapi.nat44_ei_interface_add_del_feature(
3238             sw_if_index=self.pg1.sw_if_index,
3239             is_add=1)
3240         self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=2,
3241                                                        start_port=1025,
3242                                                        end_port=1027)
3243
3244         pkts = []
3245         for port in range(0, 5):
3246             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3247                  IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3248                  TCP(sport=1125 + port))
3249             pkts.append(p)
3250         self.pg0.add_stream(pkts)
3251         self.pg_enable_capture(self.pg_interfaces)
3252         self.pg_start()
3253         capture = self.pg1.get_capture(3)
3254         for p in capture:
3255             tcp = p[TCP]
3256             self.assertGreaterEqual(tcp.sport, 1025)
3257             self.assertLessEqual(tcp.sport, 1027)
3258
3259     def test_multiple_outside_vrf(self):
3260         """ NAT44EI Multiple outside VRF """
3261         vrf_id1 = 1
3262         vrf_id2 = 2
3263
3264         self.pg1.unconfig_ip4()
3265         self.pg2.unconfig_ip4()
3266         self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
3267         self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
3268         self.pg1.set_table_ip4(vrf_id1)
3269         self.pg2.set_table_ip4(vrf_id2)
3270         self.pg1.config_ip4()
3271         self.pg2.config_ip4()
3272         self.pg1.resolve_arp()
3273         self.pg2.resolve_arp()
3274
3275         self.nat44_add_address(self.nat_addr)
3276         flags = self.config_flags.NAT44_EI_IF_INSIDE
3277         self.vapi.nat44_ei_interface_add_del_feature(
3278             sw_if_index=self.pg0.sw_if_index,
3279             flags=flags, is_add=1)
3280         self.vapi.nat44_ei_interface_add_del_feature(
3281             sw_if_index=self.pg1.sw_if_index,
3282             is_add=1)
3283         self.vapi.nat44_ei_interface_add_del_feature(
3284             sw_if_index=self.pg2.sw_if_index,
3285             is_add=1)
3286
3287         try:
3288             # first VRF
3289             pkts = self.create_stream_in(self.pg0, self.pg1)
3290             self.pg0.add_stream(pkts)
3291             self.pg_enable_capture(self.pg_interfaces)
3292             self.pg_start()
3293             capture = self.pg1.get_capture(len(pkts))
3294             self.verify_capture_out(capture, self.nat_addr)
3295
3296             pkts = self.create_stream_out(self.pg1, self.nat_addr)
3297             self.pg1.add_stream(pkts)
3298             self.pg_enable_capture(self.pg_interfaces)
3299             self.pg_start()
3300             capture = self.pg0.get_capture(len(pkts))
3301             self.verify_capture_in(capture, self.pg0)
3302
3303             self.tcp_port_in = 60303
3304             self.udp_port_in = 60304
3305             self.icmp_id_in = 60305
3306
3307             # second VRF
3308             pkts = self.create_stream_in(self.pg0, self.pg2)
3309             self.pg0.add_stream(pkts)
3310             self.pg_enable_capture(self.pg_interfaces)
3311             self.pg_start()
3312             capture = self.pg2.get_capture(len(pkts))
3313             self.verify_capture_out(capture, self.nat_addr)
3314
3315             pkts = self.create_stream_out(self.pg2, self.nat_addr)
3316             self.pg2.add_stream(pkts)
3317             self.pg_enable_capture(self.pg_interfaces)
3318             self.pg_start()
3319             capture = self.pg0.get_capture(len(pkts))
3320             self.verify_capture_in(capture, self.pg0)
3321
3322         finally:
3323             self.nat44_add_address(self.nat_addr, is_add=0)
3324             self.pg1.unconfig_ip4()
3325             self.pg2.unconfig_ip4()
3326             self.pg1.set_table_ip4(0)
3327             self.pg2.set_table_ip4(0)
3328             self.pg1.config_ip4()
3329             self.pg2.config_ip4()
3330             self.pg1.resolve_arp()
3331             self.pg2.resolve_arp()
3332
3333     def test_mss_clamping(self):
3334         """ NAT44EI TCP MSS clamping """
3335         self.nat44_add_address(self.nat_addr)
3336         flags = self.config_flags.NAT44_EI_IF_INSIDE
3337         self.vapi.nat44_ei_interface_add_del_feature(
3338             sw_if_index=self.pg0.sw_if_index,
3339             flags=flags, is_add=1)
3340         self.vapi.nat44_ei_interface_add_del_feature(
3341             sw_if_index=self.pg1.sw_if_index,
3342             is_add=1)
3343
3344         p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3345              IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3346              TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
3347                  flags="S", options=[('MSS', 1400)]))
3348
3349         self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1000)
3350         self.pg0.add_stream(p)
3351         self.pg_enable_capture(self.pg_interfaces)
3352         self.pg_start()
3353         capture = self.pg1.get_capture(1)
3354         # Negotiated MSS value greater than configured - changed
3355         self.verify_mss_value(capture[0], 1000)
3356
3357         self.vapi.nat44_ei_set_mss_clamping(enable=0, mss_value=1500)
3358         self.pg0.add_stream(p)
3359         self.pg_enable_capture(self.pg_interfaces)
3360         self.pg_start()
3361         capture = self.pg1.get_capture(1)
3362         # MSS clamping disabled - negotiated MSS unchanged
3363         self.verify_mss_value(capture[0], 1400)
3364
3365         self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1500)
3366         self.pg0.add_stream(p)
3367         self.pg_enable_capture(self.pg_interfaces)
3368         self.pg_start()
3369         capture = self.pg1.get_capture(1)
3370         # Negotiated MSS value smaller than configured - unchanged
3371         self.verify_mss_value(capture[0], 1400)
3372
3373     def test_ha_send(self):
3374         """ NAT44EI Send HA session synchronization events (active) """
3375         flags = self.config_flags.NAT44_EI_IF_INSIDE
3376         self.vapi.nat44_ei_interface_add_del_feature(
3377             sw_if_index=self.pg0.sw_if_index,
3378             flags=flags, is_add=1)
3379         self.vapi.nat44_ei_interface_add_del_feature(
3380             sw_if_index=self.pg1.sw_if_index,
3381             is_add=1)
3382         self.nat44_add_address(self.nat_addr)
3383
3384         self.vapi.nat44_ei_ha_set_listener(
3385             ip_address=self.pg3.local_ip4, port=12345, path_mtu=512)
3386         self.vapi.nat44_ei_ha_set_failover(
3387             ip_address=self.pg3.remote_ip4, port=12346,
3388             session_refresh_interval=10)
3389         bind_layers(UDP, HANATStateSync, sport=12345)
3390
3391         # create sessions
3392         pkts = self.create_stream_in(self.pg0, self.pg1)
3393         self.pg0.add_stream(pkts)
3394         self.pg_enable_capture(self.pg_interfaces)
3395         self.pg_start()
3396         capture = self.pg1.get_capture(len(pkts))
3397         self.verify_capture_out(capture)
3398         # active send HA events
3399         self.vapi.nat44_ei_ha_flush()
3400         stats = self.statistics.get_counter('/nat44-ei/ha/add-event-send')
3401         self.assertEqual(stats[0][0], 3)
3402         capture = self.pg3.get_capture(1)
3403         p = capture[0]
3404         self.assert_packet_checksums_valid(p)
3405         try:
3406             ip = p[IP]
3407             udp = p[UDP]
3408             hanat = p[HANATStateSync]
3409         except IndexError:
3410             self.logger.error(ppp("Invalid packet:", p))
3411             raise
3412         else:
3413             self.assertEqual(ip.src, self.pg3.local_ip4)
3414             self.assertEqual(ip.dst, self.pg3.remote_ip4)
3415             self.assertEqual(udp.sport, 12345)
3416             self.assertEqual(udp.dport, 12346)
3417             self.assertEqual(hanat.version, 1)
3418             self.assertEqual(hanat.thread_index, 0)
3419             self.assertEqual(hanat.count, 3)
3420             seq = hanat.sequence_number
3421             for event in hanat.events:
3422                 self.assertEqual(event.event_type, 1)
3423                 self.assertEqual(event.in_addr, self.pg0.remote_ip4)
3424                 self.assertEqual(event.out_addr, self.nat_addr)
3425                 self.assertEqual(event.fib_index, 0)
3426
3427         # ACK received events
3428         ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3429                IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3430                UDP(sport=12346, dport=12345) /
3431                HANATStateSync(sequence_number=seq, flags='ACK'))
3432         self.pg3.add_stream(ack)
3433         self.pg_start()
3434         stats = self.statistics.get_counter('/nat44-ei/ha/ack-recv')
3435         self.assertEqual(stats[0][0], 1)
3436
3437         # delete one session
3438         self.pg_enable_capture(self.pg_interfaces)
3439         self.vapi.nat44_ei_del_session(
3440             address=self.pg0.remote_ip4, port=self.tcp_port_in,
3441             protocol=IP_PROTOS.tcp, flags=self.config_flags.NAT44_EI_IF_INSIDE)
3442         self.vapi.nat44_ei_ha_flush()
3443         stats = self.statistics.get_counter('/nat44-ei/ha/del-event-send')
3444         self.assertEqual(stats[0][0], 1)
3445         capture = self.pg3.get_capture(1)
3446         p = capture[0]
3447         try:
3448             hanat = p[HANATStateSync]
3449         except IndexError:
3450             self.logger.error(ppp("Invalid packet:", p))
3451             raise
3452         else:
3453             self.assertGreater(hanat.sequence_number, seq)
3454
3455         # do not send ACK, active retry send HA event again
3456         self.pg_enable_capture(self.pg_interfaces)
3457         sleep(12)
3458         stats = self.statistics.get_counter('/nat44-ei/ha/retry-count')
3459         self.assertEqual(stats[0][0], 3)
3460         stats = self.statistics.get_counter('/nat44-ei/ha/missed-count')
3461         self.assertEqual(stats[0][0], 1)
3462         capture = self.pg3.get_capture(3)
3463         for packet in capture:
3464             self.assertEqual(packet, p)
3465
3466         # session counters refresh
3467         pkts = self.create_stream_out(self.pg1)
3468         self.pg1.add_stream(pkts)
3469         self.pg_enable_capture(self.pg_interfaces)
3470         self.pg_start()
3471         self.pg0.get_capture(2)
3472         self.vapi.nat44_ei_ha_flush()
3473         stats = self.statistics.get_counter('/nat44-ei/ha/refresh-event-send')
3474         self.assertEqual(stats[0][0], 2)
3475         capture = self.pg3.get_capture(1)
3476         p = capture[0]
3477         self.assert_packet_checksums_valid(p)
3478         try:
3479             ip = p[IP]
3480             udp = p[UDP]
3481             hanat = p[HANATStateSync]
3482         except IndexError:
3483             self.logger.error(ppp("Invalid packet:", p))
3484             raise
3485         else:
3486             self.assertEqual(ip.src, self.pg3.local_ip4)
3487             self.assertEqual(ip.dst, self.pg3.remote_ip4)
3488             self.assertEqual(udp.sport, 12345)
3489             self.assertEqual(udp.dport, 12346)
3490             self.assertEqual(hanat.version, 1)
3491             self.assertEqual(hanat.count, 2)
3492             seq = hanat.sequence_number
3493             for event in hanat.events:
3494                 self.assertEqual(event.event_type, 3)
3495                 self.assertEqual(event.out_addr, self.nat_addr)
3496                 self.assertEqual(event.fib_index, 0)
3497                 self.assertEqual(event.total_pkts, 2)
3498                 self.assertGreater(event.total_bytes, 0)
3499
3500         ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3501                IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3502                UDP(sport=12346, dport=12345) /
3503                HANATStateSync(sequence_number=seq, flags='ACK'))
3504         self.pg3.add_stream(ack)
3505         self.pg_start()
3506         stats = self.statistics.get_counter('/nat44-ei/ha/ack-recv')
3507         self.assertEqual(stats[0][0], 2)
3508
3509     def test_ha_recv(self):
3510         """ NAT44EI Receive HA session synchronization events (passive) """
3511         self.nat44_add_address(self.nat_addr)
3512         flags = self.config_flags.NAT44_EI_IF_INSIDE
3513         self.vapi.nat44_ei_interface_add_del_feature(
3514             sw_if_index=self.pg0.sw_if_index,
3515             flags=flags, is_add=1)
3516         self.vapi.nat44_ei_interface_add_del_feature(
3517             sw_if_index=self.pg1.sw_if_index,
3518             is_add=1)
3519         self.vapi.nat44_ei_ha_set_listener(ip_address=self.pg3.local_ip4,
3520                                            port=12345, path_mtu=512)
3521         bind_layers(UDP, HANATStateSync, sport=12345)
3522
3523         self.tcp_port_out = random.randint(1025, 65535)
3524         self.udp_port_out = random.randint(1025, 65535)
3525
3526         # send HA session add events to failover/passive
3527         p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3528              IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3529              UDP(sport=12346, dport=12345) /
3530              HANATStateSync(sequence_number=1, events=[
3531                  Event(event_type='add', protocol='tcp',
3532                        in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3533                        in_port=self.tcp_port_in, out_port=self.tcp_port_out,
3534                        eh_addr=self.pg1.remote_ip4,
3535                        ehn_addr=self.pg1.remote_ip4,
3536                        eh_port=self.tcp_external_port,
3537                        ehn_port=self.tcp_external_port, fib_index=0),
3538                  Event(event_type='add', protocol='udp',
3539                        in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3540                        in_port=self.udp_port_in, out_port=self.udp_port_out,
3541                        eh_addr=self.pg1.remote_ip4,
3542                        ehn_addr=self.pg1.remote_ip4,
3543                        eh_port=self.udp_external_port,
3544                        ehn_port=self.udp_external_port, fib_index=0)]))
3545
3546         self.pg3.add_stream(p)
3547         self.pg_enable_capture(self.pg_interfaces)
3548         self.pg_start()
3549         # receive ACK
3550         capture = self.pg3.get_capture(1)
3551         p = capture[0]
3552         try:
3553             hanat = p[HANATStateSync]
3554         except IndexError:
3555             self.logger.error(ppp("Invalid packet:", p))
3556             raise
3557         else:
3558             self.assertEqual(hanat.sequence_number, 1)
3559             self.assertEqual(hanat.flags, 'ACK')
3560             self.assertEqual(hanat.version, 1)
3561             self.assertEqual(hanat.thread_index, 0)
3562         stats = self.statistics.get_counter('/nat44-ei/ha/ack-send')
3563         self.assertEqual(stats[0][0], 1)
3564         stats = self.statistics.get_counter('/nat44-ei/ha/add-event-recv')
3565         self.assertEqual(stats[0][0], 2)
3566         users = self.statistics.get_counter('/nat44-ei/total-users')
3567         self.assertEqual(users[0][0], 1)
3568         sessions = self.statistics.get_counter('/nat44-ei/total-sessions')
3569         self.assertEqual(sessions[0][0], 2)
3570         users = self.vapi.nat44_ei_user_dump()
3571         self.assertEqual(len(users), 1)
3572         self.assertEqual(str(users[0].ip_address),
3573                          self.pg0.remote_ip4)
3574         # there should be 2 sessions created by HA
3575         sessions = self.vapi.nat44_ei_user_session_dump(
3576             users[0].ip_address, users[0].vrf_id)
3577         self.assertEqual(len(sessions), 2)
3578         for session in sessions:
3579             self.assertEqual(str(session.inside_ip_address),
3580                              self.pg0.remote_ip4)
3581             self.assertEqual(str(session.outside_ip_address),
3582                              self.nat_addr)
3583             self.assertIn(session.inside_port,
3584                           [self.tcp_port_in, self.udp_port_in])
3585             self.assertIn(session.outside_port,
3586                           [self.tcp_port_out, self.udp_port_out])
3587             self.assertIn(session.protocol, [IP_PROTOS.tcp, IP_PROTOS.udp])
3588
3589         # send HA session delete event to failover/passive
3590         p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3591              IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3592              UDP(sport=12346, dport=12345) /
3593              HANATStateSync(sequence_number=2, events=[
3594                  Event(event_type='del', protocol='udp',
3595                        in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3596                        in_port=self.udp_port_in, out_port=self.udp_port_out,
3597                        eh_addr=self.pg1.remote_ip4,
3598                        ehn_addr=self.pg1.remote_ip4,
3599                        eh_port=self.udp_external_port,
3600                        ehn_port=self.udp_external_port, fib_index=0)]))
3601
3602         self.pg3.add_stream(p)
3603         self.pg_enable_capture(self.pg_interfaces)
3604         self.pg_start()
3605         # receive ACK
3606         capture = self.pg3.get_capture(1)
3607         p = capture[0]
3608         try:
3609             hanat = p[HANATStateSync]
3610         except IndexError:
3611             self.logger.error(ppp("Invalid packet:", p))
3612             raise
3613         else:
3614             self.assertEqual(hanat.sequence_number, 2)
3615             self.assertEqual(hanat.flags, 'ACK')
3616             self.assertEqual(hanat.version, 1)
3617         users = self.vapi.nat44_ei_user_dump()
3618         self.assertEqual(len(users), 1)
3619         self.assertEqual(str(users[0].ip_address),
3620                          self.pg0.remote_ip4)
3621         # now we should have only 1 session, 1 deleted by HA
3622         sessions = self.vapi.nat44_ei_user_session_dump(users[0].ip_address,
3623                                                         users[0].vrf_id)
3624         self.assertEqual(len(sessions), 1)
3625         stats = self.statistics.get_counter('/nat44-ei/ha/del-event-recv')
3626         self.assertEqual(stats[0][0], 1)
3627
3628         stats = self.statistics.get_err_counter(
3629             '/err/nat44-ei-ha/pkts-processed')
3630         self.assertEqual(stats, 2)
3631
3632         # send HA session refresh event to failover/passive
3633         p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3634              IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3635              UDP(sport=12346, dport=12345) /
3636              HANATStateSync(sequence_number=3, events=[
3637                  Event(event_type='refresh', protocol='tcp',
3638                        in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3639                        in_port=self.tcp_port_in, out_port=self.tcp_port_out,
3640                        eh_addr=self.pg1.remote_ip4,
3641                        ehn_addr=self.pg1.remote_ip4,
3642                        eh_port=self.tcp_external_port,
3643                        ehn_port=self.tcp_external_port, fib_index=0,
3644                        total_bytes=1024, total_pkts=2)]))
3645         self.pg3.add_stream(p)
3646         self.pg_enable_capture(self.pg_interfaces)
3647         self.pg_start()
3648         # receive ACK
3649         capture = self.pg3.get_capture(1)
3650         p = capture[0]
3651         try:
3652             hanat = p[HANATStateSync]
3653         except IndexError:
3654             self.logger.error(ppp("Invalid packet:", p))
3655             raise
3656         else:
3657             self.assertEqual(hanat.sequence_number, 3)
3658             self.assertEqual(hanat.flags, 'ACK')
3659             self.assertEqual(hanat.version, 1)
3660         users = self.vapi.nat44_ei_user_dump()
3661         self.assertEqual(len(users), 1)
3662         self.assertEqual(str(users[0].ip_address),
3663                          self.pg0.remote_ip4)
3664         sessions = self.vapi.nat44_ei_user_session_dump(
3665             users[0].ip_address, users[0].vrf_id)
3666         self.assertEqual(len(sessions), 1)
3667         session = sessions[0]
3668         self.assertEqual(session.total_bytes, 1024)
3669         self.assertEqual(session.total_pkts, 2)
3670         stats = self.statistics.get_counter(
3671             '/nat44-ei/ha/refresh-event-recv')
3672         self.assertEqual(stats[0][0], 1)
3673
3674         stats = self.statistics.get_err_counter(
3675             '/err/nat44-ei-ha/pkts-processed')
3676         self.assertEqual(stats, 3)
3677
3678         # send packet to test session created by HA
3679         p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
3680              IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3681              TCP(sport=self.tcp_external_port, dport=self.tcp_port_out))
3682         self.pg1.add_stream(p)
3683         self.pg_enable_capture(self.pg_interfaces)
3684         self.pg_start()
3685         capture = self.pg0.get_capture(1)
3686         p = capture[0]
3687         try:
3688             ip = p[IP]
3689             tcp = p[TCP]
3690         except IndexError:
3691             self.logger.error(ppp("Invalid packet:", p))
3692             raise
3693         else:
3694             self.assertEqual(ip.src, self.pg1.remote_ip4)
3695             self.assertEqual(ip.dst, self.pg0.remote_ip4)
3696             self.assertEqual(tcp.sport, self.tcp_external_port)
3697             self.assertEqual(tcp.dport, self.tcp_port_in)
3698
3699     def reconfigure_frame_queue_nelts(self, frame_queue_nelts):
3700         self.vapi.nat44_ei_plugin_enable_disable(enable=0)
3701         self.vapi.nat44_ei_set_fq_options(frame_queue_nelts=frame_queue_nelts)
3702         # keep plugin configuration persistent
3703         self.plugin_enable()
3704         return self.vapi.nat44_ei_show_fq_options().frame_queue_nelts
3705
3706     def test_set_frame_queue_nelts(self):
3707         """ NAT44 EI API test - worker handoff frame queue elements """
3708         self.assertEqual(self.reconfigure_frame_queue_nelts(512), 512)
3709
3710     def show_commands_at_teardown(self):
3711         self.logger.info(self.vapi.cli("show nat44 ei timeouts"))
3712         self.logger.info(self.vapi.cli("show nat44 ei addresses"))
3713         self.logger.info(self.vapi.cli("show nat44 ei interfaces"))
3714         self.logger.info(self.vapi.cli("show nat44 ei static mappings"))
3715         self.logger.info(self.vapi.cli("show nat44 ei interface address"))
3716         self.logger.info(self.vapi.cli("show nat44 ei sessions detail"))
3717         self.logger.info(self.vapi.cli("show nat44 ei hash tables detail"))
3718         self.logger.info(self.vapi.cli("show nat44 ei ha"))
3719         self.logger.info(
3720             self.vapi.cli("show nat44 ei addr-port-assignment-alg"))
3721
3722     def test_outside_address_distribution(self):
3723         """ Outside address distribution based on source address """
3724
3725         x = 100
3726         nat_addresses = []
3727
3728         for i in range(1, x):
3729             a = "10.0.0.%d" % i
3730             nat_addresses.append(a)
3731
3732         flags = self.config_flags.NAT44_EI_IF_INSIDE
3733         self.vapi.nat44_ei_interface_add_del_feature(
3734             sw_if_index=self.pg0.sw_if_index,
3735             flags=flags, is_add=1)
3736         self.vapi.nat44_ei_interface_add_del_feature(
3737             sw_if_index=self.pg1.sw_if_index,
3738             is_add=1)
3739
3740         self.vapi.nat44_ei_add_del_address_range(
3741             first_ip_address=nat_addresses[0],
3742             last_ip_address=nat_addresses[-1],
3743             vrf_id=0xFFFFFFFF, is_add=1)
3744
3745         self.pg0.generate_remote_hosts(x)
3746
3747         pkts = []
3748         for i in range(x):
3749             p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3750                  IP(src=self.pg0.remote_hosts[i].ip4,
3751                      dst=self.pg1.remote_ip4) /
3752                  UDP(sport=7000+i, dport=80+i))
3753             pkts.append(p)
3754
3755         self.pg0.add_stream(pkts)
3756         self.pg_enable_capture(self.pg_interfaces)
3757         self.pg_start()
3758         recvd = self.pg1.get_capture(len(pkts))
3759         for (p_sent, p_recvd) in zip(pkts, recvd):
3760             packed = socket.inet_aton(p_sent[IP].src)
3761             numeric = struct.unpack("!L", packed)[0]
3762             numeric = socket.htonl(numeric)
3763             a = nat_addresses[(numeric-1) % len(nat_addresses)]
3764             self.assertEqual(a, p_recvd[IP].src, "Packet not translated")
3765
3766
3767 class TestNAT44Out2InDPO(MethodHolder):
3768     """ NAT44EI Test Cases using out2in DPO """
3769
3770     @classmethod
3771     def setUpClass(cls):
3772         super(TestNAT44Out2InDPO, cls).setUpClass()
3773         cls.vapi.cli("set log class nat44-ei level debug")
3774
3775         cls.tcp_port_in = 6303
3776         cls.tcp_port_out = 6303
3777         cls.udp_port_in = 6304
3778         cls.udp_port_out = 6304
3779         cls.icmp_id_in = 6305
3780         cls.icmp_id_out = 6305
3781         cls.nat_addr = '10.0.0.3'
3782         cls.dst_ip4 = '192.168.70.1'
3783
3784         cls.create_pg_interfaces(range(2))
3785
3786         cls.pg0.admin_up()
3787         cls.pg0.config_ip4()
3788         cls.pg0.resolve_arp()
3789
3790         cls.pg1.admin_up()
3791         cls.pg1.config_ip6()
3792         cls.pg1.resolve_ndp()
3793
3794         r1 = VppIpRoute(cls, "::", 0,
3795                         [VppRoutePath(cls.pg1.remote_ip6,
3796                                       cls.pg1.sw_if_index)],
3797                         register=False)
3798         r1.add_vpp_config()
3799
3800     def setUp(self):
3801         super(TestNAT44Out2InDPO, self).setUp()
3802         flags = self.config_flags.NAT44_EI_OUT2IN_DPO
3803         self.vapi.nat44_ei_plugin_enable_disable(enable=1, flags=flags)
3804
3805     def tearDown(self):
3806         super(TestNAT44Out2InDPO, self).tearDown()
3807         if not self.vpp_dead:
3808             self.vapi.nat44_ei_plugin_enable_disable(enable=0)
3809             self.vapi.cli("clear logging")
3810
3811     def configure_xlat(self):
3812         self.dst_ip6_pfx = '1:2:3::'
3813         self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3814                                               self.dst_ip6_pfx)
3815         self.dst_ip6_pfx_len = 96
3816         self.src_ip6_pfx = '4:5:6::'
3817         self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3818                                               self.src_ip6_pfx)
3819         self.src_ip6_pfx_len = 96
3820         self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
3821                                  self.src_ip6_pfx_n, self.src_ip6_pfx_len,
3822                                  '\x00\x00\x00\x00', 0)
3823
3824     @unittest.skip('Temporary disabled')
3825     def test_464xlat_ce(self):
3826         """ Test 464XLAT CE with NAT44EI """
3827
3828         self.configure_xlat()
3829
3830         flags = self.config_flags.NAT44_EI_IF_INSIDE
3831         self.vapi.nat44_ei_interface_add_del_feature(
3832             sw_if_index=self.pg0.sw_if_index,
3833             flags=flags, is_add=1)
3834         self.vapi.nat44_ei_add_del_address_range(
3835             first_ip_address=self.nat_addr_n,
3836             last_ip_address=self.nat_addr_n,
3837             vrf_id=0xFFFFFFFF, is_add=1)
3838
3839         out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3840                                        self.dst_ip6_pfx_len)
3841         out_dst_ip6 = self.compose_ip6(self.nat_addr, self.src_ip6_pfx,
3842                                        self.src_ip6_pfx_len)
3843
3844         try:
3845             pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3846             self.pg0.add_stream(pkts)
3847             self.pg_enable_capture(self.pg_interfaces)
3848             self.pg_start()
3849             capture = self.pg1.get_capture(len(pkts))
3850             self.verify_capture_out_ip6(capture, nat_ip=out_dst_ip6,
3851                                         dst_ip=out_src_ip6)
3852
3853             pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6,
3854                                               out_dst_ip6)
3855             self.pg1.add_stream(pkts)
3856             self.pg_enable_capture(self.pg_interfaces)
3857             self.pg_start()
3858             capture = self.pg0.get_capture(len(pkts))
3859             self.verify_capture_in(capture, self.pg0)
3860         finally:
3861             self.vapi.nat44_ei_interface_add_del_feature(
3862                 sw_if_index=self.pg0.sw_if_index,
3863                 flags=flags)
3864             self.vapi.nat44_ei_add_del_address_range(
3865                 first_ip_address=self.nat_addr_n,
3866                 last_ip_address=self.nat_addr_n,
3867                 vrf_id=0xFFFFFFFF)
3868
3869     @unittest.skip('Temporary disabled')
3870     def test_464xlat_ce_no_nat(self):
3871         """ Test 464XLAT CE without NAT44EI """
3872
3873         self.configure_xlat()
3874
3875         out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3876                                        self.dst_ip6_pfx_len)
3877         out_dst_ip6 = self.compose_ip6(self.pg0.remote_ip4, self.src_ip6_pfx,
3878                                        self.src_ip6_pfx_len)
3879
3880         pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3881         self.pg0.add_stream(pkts)
3882         self.pg_enable_capture(self.pg_interfaces)
3883         self.pg_start()
3884         capture = self.pg1.get_capture(len(pkts))
3885         self.verify_capture_out_ip6(capture, dst_ip=out_src_ip6,
3886                                     nat_ip=out_dst_ip6, same_port=True)
3887
3888         pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6, out_dst_ip6)
3889         self.pg1.add_stream(pkts)
3890         self.pg_enable_capture(self.pg_interfaces)
3891         self.pg_start()
3892         capture = self.pg0.get_capture(len(pkts))
3893         self.verify_capture_in(capture, self.pg0)
3894
3895
3896 class TestNAT44EIMW(MethodHolder):
3897     """ NAT44EI Test Cases (multiple workers) """
3898
3899     worker_config = "workers %d" % 2
3900
3901     max_translations = 10240
3902     max_users = 10240
3903
3904     @classmethod
3905     def setUpClass(cls):
3906         super(TestNAT44EIMW, cls).setUpClass()
3907         cls.vapi.cli("set log class nat level debug")
3908
3909         cls.tcp_port_in = 6303
3910         cls.tcp_port_out = 6303
3911         cls.udp_port_in = 6304
3912         cls.udp_port_out = 6304
3913         cls.icmp_id_in = 6305
3914         cls.icmp_id_out = 6305
3915         cls.nat_addr = '10.0.0.3'
3916         cls.ipfix_src_port = 4739
3917         cls.ipfix_domain_id = 1
3918         cls.tcp_external_port = 80
3919         cls.udp_external_port = 69
3920
3921         cls.create_pg_interfaces(range(10))
3922         cls.interfaces = list(cls.pg_interfaces[0:4])
3923
3924         for i in cls.interfaces:
3925             i.admin_up()
3926             i.config_ip4()
3927             i.resolve_arp()
3928
3929         cls.pg0.generate_remote_hosts(3)
3930         cls.pg0.configure_ipv4_neighbors()
3931
3932         cls.pg1.generate_remote_hosts(1)
3933         cls.pg1.configure_ipv4_neighbors()
3934
3935         cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
3936         cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
3937         cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
3938
3939         cls.pg4._local_ip4 = "172.16.255.1"
3940         cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
3941         cls.pg4.set_table_ip4(10)
3942         cls.pg5._local_ip4 = "172.17.255.3"
3943         cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
3944         cls.pg5.set_table_ip4(10)
3945         cls.pg6._local_ip4 = "172.16.255.1"
3946         cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
3947         cls.pg6.set_table_ip4(20)
3948         for i in cls.overlapping_interfaces:
3949             i.config_ip4()
3950             i.admin_up()
3951             i.resolve_arp()
3952
3953         cls.pg7.admin_up()
3954         cls.pg8.admin_up()
3955
3956         cls.pg9.generate_remote_hosts(2)
3957         cls.pg9.config_ip4()
3958         cls.vapi.sw_interface_add_del_address(
3959             sw_if_index=cls.pg9.sw_if_index,
3960             prefix="10.0.0.1/24")
3961
3962         cls.pg9.admin_up()
3963         cls.pg9.resolve_arp()
3964         cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
3965         cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
3966         cls.pg9.resolve_arp()
3967
3968     def setUp(self):
3969         super(TestNAT44EIMW, self).setUp()
3970         self.vapi.nat44_ei_plugin_enable_disable(
3971             sessions=self.max_translations,
3972             users=self.max_users, enable=1)
3973
3974     def tearDown(self):
3975         super(TestNAT44EIMW, self).tearDown()
3976         if not self.vpp_dead:
3977             self.vapi.nat44_ei_ipfix_enable_disable(
3978                 domain_id=self.ipfix_domain_id,
3979                 src_port=self.ipfix_src_port,
3980                 enable=0)
3981             self.ipfix_src_port = 4739
3982             self.ipfix_domain_id = 1
3983
3984             self.vapi.nat44_ei_plugin_enable_disable(enable=0)
3985             self.vapi.cli("clear logging")
3986
3987     def test_hairpinning(self):
3988         """ NAT44EI hairpinning - 1:1 NAPT """
3989
3990         host = self.pg0.remote_hosts[0]
3991         server = self.pg0.remote_hosts[1]
3992         host_in_port = 1234
3993         host_out_port = 0
3994         server_in_port = 5678
3995         server_out_port = 8765
3996         worker_1 = 1
3997         worker_2 = 2
3998
3999         self.nat44_add_address(self.nat_addr)
4000         flags = self.config_flags.NAT44_EI_IF_INSIDE
4001         self.vapi.nat44_ei_interface_add_del_feature(
4002             sw_if_index=self.pg0.sw_if_index,
4003             flags=flags, is_add=1)
4004         self.vapi.nat44_ei_interface_add_del_feature(
4005             sw_if_index=self.pg1.sw_if_index,
4006             is_add=1)
4007
4008         # add static mapping for server
4009         self.nat44_add_static_mapping(server.ip4, self.nat_addr,
4010                                       server_in_port, server_out_port,
4011                                       proto=IP_PROTOS.tcp)
4012
4013         cnt = self.statistics.get_counter('/nat44-ei/hairpinning')
4014         # send packet from host to server
4015         p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
4016              IP(src=host.ip4, dst=self.nat_addr) /
4017              TCP(sport=host_in_port, dport=server_out_port))
4018         self.pg0.add_stream(p)
4019         self.pg_enable_capture(self.pg_interfaces)
4020         self.pg_start()
4021         capture = self.pg0.get_capture(1)
4022         p = capture[0]
4023         try:
4024             ip = p[IP]
4025             tcp = p[TCP]
4026             self.assertEqual(ip.src, self.nat_addr)
4027             self.assertEqual(ip.dst, server.ip4)
4028             self.assertNotEqual(tcp.sport, host_in_port)
4029             self.assertEqual(tcp.dport, server_in_port)
4030             self.assert_packet_checksums_valid(p)
4031             host_out_port = tcp.sport
4032         except:
4033             self.logger.error(ppp("Unexpected or invalid packet:", p))
4034             raise
4035
4036         after = self.statistics.get_counter('/nat44-ei/hairpinning')
4037
4038         if_idx = self.pg0.sw_if_index
4039         self.assertEqual(after[worker_2][if_idx] - cnt[worker_1][if_idx], 1)
4040
4041         # send reply from server to host
4042         p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
4043              IP(src=server.ip4, dst=self.nat_addr) /
4044              TCP(sport=server_in_port, dport=host_out_port))
4045         self.pg0.add_stream(p)
4046         self.pg_enable_capture(self.pg_interfaces)
4047         self.pg_start()
4048         capture = self.pg0.get_capture(1)
4049         p = capture[0]
4050         try:
4051             ip = p[IP]
4052             tcp = p[TCP]
4053             self.assertEqual(ip.src, self.nat_addr)
4054             self.assertEqual(ip.dst, host.ip4)
4055             self.assertEqual(tcp.sport, server_out_port)
4056             self.assertEqual(tcp.dport, host_in_port)
4057             self.assert_packet_checksums_valid(p)
4058         except:
4059             self.logger.error(ppp("Unexpected or invalid packet:", p))
4060             raise
4061
4062         after = self.statistics.get_counter('/nat44-ei/hairpinning')
4063         if_idx = self.pg0.sw_if_index
4064         self.assertEqual(after[worker_1][if_idx] - cnt[worker_1][if_idx], 1)
4065         self.assertEqual(after[worker_2][if_idx] - cnt[worker_2][if_idx], 2)
4066
4067     def test_hairpinning2(self):
4068         """ NAT44EI hairpinning - 1:1 NAT"""
4069
4070         server1_nat_ip = "10.0.0.10"
4071         server2_nat_ip = "10.0.0.11"
4072         host = self.pg0.remote_hosts[0]
4073         server1 = self.pg0.remote_hosts[1]
4074         server2 = self.pg0.remote_hosts[2]
4075         server_tcp_port = 22
4076         server_udp_port = 20
4077
4078         self.nat44_add_address(self.nat_addr)
4079         flags = self.config_flags.NAT44_EI_IF_INSIDE
4080         self.vapi.nat44_ei_interface_add_del_feature(
4081             sw_if_index=self.pg0.sw_if_index,
4082             flags=flags, is_add=1)
4083         self.vapi.nat44_ei_interface_add_del_feature(
4084             sw_if_index=self.pg1.sw_if_index,
4085             is_add=1)
4086
4087         # add static mapping for servers
4088         self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
4089         self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
4090
4091         # host to server1
4092         pkts = []
4093         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4094              IP(src=host.ip4, dst=server1_nat_ip) /
4095              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
4096         pkts.append(p)
4097         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4098              IP(src=host.ip4, dst=server1_nat_ip) /
4099              UDP(sport=self.udp_port_in, dport=server_udp_port))
4100         pkts.append(p)
4101         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4102              IP(src=host.ip4, dst=server1_nat_ip) /
4103              ICMP(id=self.icmp_id_in, type='echo-request'))
4104         pkts.append(p)
4105         self.pg0.add_stream(pkts)
4106         self.pg_enable_capture(self.pg_interfaces)
4107         self.pg_start()
4108         capture = self.pg0.get_capture(len(pkts))
4109         for packet in capture:
4110             try:
4111                 self.assertEqual(packet[IP].src, self.nat_addr)
4112                 self.assertEqual(packet[IP].dst, server1.ip4)
4113                 if packet.haslayer(TCP):
4114                     self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
4115                     self.assertEqual(packet[TCP].dport, server_tcp_port)
4116                     self.tcp_port_out = packet[TCP].sport
4117                     self.assert_packet_checksums_valid(packet)
4118                 elif packet.haslayer(UDP):
4119                     self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
4120                     self.assertEqual(packet[UDP].dport, server_udp_port)
4121                     self.udp_port_out = packet[UDP].sport
4122                 else:
4123                     self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
4124                     self.icmp_id_out = packet[ICMP].id
4125             except:
4126                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4127                 raise
4128
4129         # server1 to host
4130         pkts = []
4131         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4132              IP(src=server1.ip4, dst=self.nat_addr) /
4133              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
4134         pkts.append(p)
4135         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4136              IP(src=server1.ip4, dst=self.nat_addr) /
4137              UDP(sport=server_udp_port, dport=self.udp_port_out))
4138         pkts.append(p)
4139         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4140              IP(src=server1.ip4, dst=self.nat_addr) /
4141              ICMP(id=self.icmp_id_out, type='echo-reply'))
4142         pkts.append(p)
4143         self.pg0.add_stream(pkts)
4144         self.pg_enable_capture(self.pg_interfaces)
4145         self.pg_start()
4146         capture = self.pg0.get_capture(len(pkts))
4147         for packet in capture:
4148             try:
4149                 self.assertEqual(packet[IP].src, server1_nat_ip)
4150                 self.assertEqual(packet[IP].dst, host.ip4)
4151                 if packet.haslayer(TCP):
4152                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
4153                     self.assertEqual(packet[TCP].sport, server_tcp_port)
4154                     self.assert_packet_checksums_valid(packet)
4155                 elif packet.haslayer(UDP):
4156                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
4157                     self.assertEqual(packet[UDP].sport, server_udp_port)
4158                 else:
4159                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4160             except:
4161                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4162                 raise
4163
4164         # server2 to server1
4165         pkts = []
4166         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4167              IP(src=server2.ip4, dst=server1_nat_ip) /
4168              TCP(sport=self.tcp_port_in, dport=server_tcp_port))
4169         pkts.append(p)
4170         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4171              IP(src=server2.ip4, dst=server1_nat_ip) /
4172              UDP(sport=self.udp_port_in, dport=server_udp_port))
4173         pkts.append(p)
4174         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4175              IP(src=server2.ip4, dst=server1_nat_ip) /
4176              ICMP(id=self.icmp_id_in, type='echo-request'))
4177         pkts.append(p)
4178         self.pg0.add_stream(pkts)
4179         self.pg_enable_capture(self.pg_interfaces)
4180         self.pg_start()
4181         capture = self.pg0.get_capture(len(pkts))
4182         for packet in capture:
4183             try:
4184                 self.assertEqual(packet[IP].src, server2_nat_ip)
4185                 self.assertEqual(packet[IP].dst, server1.ip4)
4186                 if packet.haslayer(TCP):
4187                     self.assertEqual(packet[TCP].sport, self.tcp_port_in)
4188                     self.assertEqual(packet[TCP].dport, server_tcp_port)
4189                     self.tcp_port_out = packet[TCP].sport
4190                     self.assert_packet_checksums_valid(packet)
4191                 elif packet.haslayer(UDP):
4192                     self.assertEqual(packet[UDP].sport, self.udp_port_in)
4193                     self.assertEqual(packet[UDP].dport, server_udp_port)
4194                     self.udp_port_out = packet[UDP].sport
4195                 else:
4196                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4197                     self.icmp_id_out = packet[ICMP].id
4198             except:
4199                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4200                 raise
4201
4202         # server1 to server2
4203         pkts = []
4204         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4205              IP(src=server1.ip4, dst=server2_nat_ip) /
4206              TCP(sport=server_tcp_port, dport=self.tcp_port_out))
4207         pkts.append(p)
4208         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4209              IP(src=server1.ip4, dst=server2_nat_ip) /
4210              UDP(sport=server_udp_port, dport=self.udp_port_out))
4211         pkts.append(p)
4212         p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4213              IP(src=server1.ip4, dst=server2_nat_ip) /
4214              ICMP(id=self.icmp_id_out, type='echo-reply'))
4215         pkts.append(p)
4216         self.pg0.add_stream(pkts)
4217         self.pg_enable_capture(self.pg_interfaces)
4218         self.pg_start()
4219         capture = self.pg0.get_capture(len(pkts))
4220         for packet in capture:
4221             try:
4222                 self.assertEqual(packet[IP].src, server1_nat_ip)
4223                 self.assertEqual(packet[IP].dst, server2.ip4)
4224                 if packet.haslayer(TCP):
4225                     self.assertEqual(packet[TCP].dport, self.tcp_port_in)
4226                     self.assertEqual(packet[TCP].sport, server_tcp_port)
4227                     self.assert_packet_checksums_valid(packet)
4228                 elif packet.haslayer(UDP):
4229                     self.assertEqual(packet[UDP].dport, self.udp_port_in)
4230                     self.assertEqual(packet[UDP].sport, server_udp_port)
4231                 else:
4232                     self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4233             except:
4234                 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4235                 raise
4236
4237
4238 if __name__ == '__main__':
4239     unittest.main(testRunner=VppTestRunner)