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