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