12 from framework import VppTestCase, VppTestRunner
13 from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
14 from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
15 IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
17 from scapy.data import IP_PROTOS
18 from scapy.layers.inet import IP, TCP, UDP, ICMP
19 from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
20 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
21 from scapy.layers.l2 import Ether, ARP, GRE
22 from scapy.packet import Raw
23 from syslog_rfc5424_parser import SyslogMessage, ParseError
24 from syslog_rfc5424_parser.constants import SyslogSeverity
26 from vpp_ip_route import VppIpRoute, VppRoutePath
27 from vpp_neighbor import VppNeighbor
28 from vpp_papi import VppEnum
31 # NAT HA protocol event data
34 fields_desc = [ByteEnumField("event_type", None,
35 {1: "add", 2: "del", 3: "refresh"}),
36 ByteEnumField("protocol", None,
37 {0: "other", 1: "udp", 2: "tcp", 3: "icmp"}),
38 ShortField("flags", 0),
39 IPField("in_addr", None),
40 IPField("out_addr", None),
41 ShortField("in_port", None),
42 ShortField("out_port", None),
43 IPField("eh_addr", None),
44 IPField("ehn_addr", None),
45 ShortField("eh_port", None),
46 ShortField("ehn_port", None),
47 IntField("fib_index", None),
48 IntField("total_pkts", 0),
49 LongField("total_bytes", 0)]
51 def extract_padding(self, s):
55 # NAT HA protocol header
56 class HANATStateSync(Packet):
57 name = "HA NAT state sync"
58 fields_desc = [XByteField("version", 1),
59 FlagsField("flags", 0, 8, ['ACK']),
60 FieldLenField("count", None, count_of="events"),
61 IntField("sequence_number", 1),
62 IntField("thread_index", 0),
63 PacketListField("events", [], Event,
64 count_from=lambda pkt: pkt.count)]
67 class MethodHolder(VppTestCase):
68 """ NAT create capture and verify method holder """
71 def config_flags(self):
72 return VppEnum.vl_api_nat44_ei_config_flags_t
75 def SYSLOG_SEVERITY(self):
76 return VppEnum.vl_api_syslog_severity_t
78 def nat44_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
79 local_port=0, external_port=0, vrf_id=0,
80 is_add=1, external_sw_if_index=0xFFFFFFFF,
81 proto=0, tag="", flags=0):
83 Add/delete NAT44EI static mapping
85 :param local_ip: Local IP address
86 :param external_ip: External IP address
87 :param local_port: Local port number (Optional)
88 :param external_port: External port number (Optional)
89 :param vrf_id: VRF ID (Default 0)
90 :param is_add: 1 if add, 0 if delete (Default add)
91 :param external_sw_if_index: External interface instead of IP address
92 :param proto: IP protocol (Mandatory if port specified)
93 :param tag: Opaque string tag
94 :param flags: NAT configuration flags
97 if not (local_port and external_port):
98 flags |= self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
100 self.vapi.nat44_ei_add_del_static_mapping(
102 local_ip_address=local_ip,
103 external_ip_address=external_ip,
104 external_sw_if_index=external_sw_if_index,
105 local_port=local_port,
106 external_port=external_port,
107 vrf_id=vrf_id, protocol=proto,
111 def nat44_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
113 Add/delete NAT44EI address
115 :param ip: IP address
116 :param is_add: 1 if add, 0 if delete (Default add)
118 self.vapi.nat44_ei_add_del_address_range(first_ip_address=ip,
123 def create_routes_and_neigbors(self):
124 r1 = VppIpRoute(self, self.pg7.remote_ip4, 32,
125 [VppRoutePath(self.pg7.remote_ip4,
126 self.pg7.sw_if_index)])
127 r2 = VppIpRoute(self, self.pg8.remote_ip4, 32,
128 [VppRoutePath(self.pg8.remote_ip4,
129 self.pg8.sw_if_index)])
133 n1 = VppNeighbor(self,
134 self.pg7.sw_if_index,
138 n2 = VppNeighbor(self,
139 self.pg8.sw_if_index,
146 def create_stream_in(self, in_if, out_if, dst_ip=None, ttl=64):
148 Create packet stream for inside network
150 :param in_if: Inside interface
151 :param out_if: Outside interface
152 :param dst_ip: Destination address
153 :param ttl: TTL of generated packets
156 dst_ip = out_if.remote_ip4
160 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
161 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
162 TCP(sport=self.tcp_port_in, dport=20))
166 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
167 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
168 UDP(sport=self.udp_port_in, dport=20))
172 p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
173 IP(src=in_if.remote_ip4, dst=dst_ip, ttl=ttl) /
174 ICMP(id=self.icmp_id_in, type='echo-request'))
179 def compose_ip6(self, ip4, pref, plen):
181 Compose IPv4-embedded IPv6 addresses
183 :param ip4: IPv4 address
184 :param pref: IPv6 prefix
185 :param plen: IPv6 prefix length
186 :returns: IPv4-embedded IPv6 addresses
188 pref_n = list(socket.inet_pton(socket.AF_INET6, pref))
189 ip4_n = list(socket.inet_pton(socket.AF_INET, ip4))
204 pref_n[10] = ip4_n[3]
208 pref_n[10] = ip4_n[2]
209 pref_n[11] = ip4_n[3]
212 pref_n[10] = ip4_n[1]
213 pref_n[11] = ip4_n[2]
214 pref_n[12] = ip4_n[3]
216 pref_n[12] = ip4_n[0]
217 pref_n[13] = ip4_n[1]
218 pref_n[14] = ip4_n[2]
219 pref_n[15] = ip4_n[3]
220 packed_pref_n = b''.join([scapy.compat.chb(x) for x in pref_n])
221 return socket.inet_ntop(socket.AF_INET6, packed_pref_n)
223 def create_stream_out(self, out_if, dst_ip=None, ttl=64,
224 use_inside_ports=False):
226 Create packet stream for outside network
228 :param out_if: Outside interface
229 :param dst_ip: Destination IP address (Default use global NAT address)
230 :param ttl: TTL of generated packets
231 :param use_inside_ports: Use inside NAT ports as destination ports
232 instead of outside ports
235 dst_ip = self.nat_addr
236 if not use_inside_ports:
237 tcp_port = self.tcp_port_out
238 udp_port = self.udp_port_out
239 icmp_id = self.icmp_id_out
241 tcp_port = self.tcp_port_in
242 udp_port = self.udp_port_in
243 icmp_id = self.icmp_id_in
246 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
247 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
248 TCP(dport=tcp_port, sport=20))
252 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
253 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
254 UDP(dport=udp_port, sport=20))
258 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
259 IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
260 ICMP(id=icmp_id, type='echo-reply'))
265 def create_stream_out_ip6(self, out_if, src_ip, dst_ip, hl=64):
267 Create packet stream for outside network
269 :param out_if: Outside interface
270 :param dst_ip: Destination IP address (Default use global NAT address)
271 :param hl: HL of generated packets
275 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
276 IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
277 TCP(dport=self.tcp_port_out, sport=20))
281 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
282 IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
283 UDP(dport=self.udp_port_out, sport=20))
287 p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
288 IPv6(src=src_ip, dst=dst_ip, hlim=hl) /
289 ICMPv6EchoReply(id=self.icmp_id_out))
294 def verify_capture_out(self, capture, nat_ip=None, same_port=False,
295 dst_ip=None, is_ip6=False, ignore_port=False):
297 Verify captured packets on outside network
299 :param capture: Captured packets
300 :param nat_ip: Translated IP address (Default use global NAT address)
301 :param same_port: Source port number is not translated (Default False)
302 :param dst_ip: Destination IP address (Default do not verify)
303 :param is_ip6: If L3 protocol is IPv6 (Default False)
307 ICMP46 = ICMPv6EchoRequest
312 nat_ip = self.nat_addr
313 for packet in capture:
316 self.assert_packet_checksums_valid(packet)
317 self.assertEqual(packet[IP46].src, nat_ip)
318 if dst_ip is not None:
319 self.assertEqual(packet[IP46].dst, dst_ip)
320 if packet.haslayer(TCP):
324 packet[TCP].sport, self.tcp_port_in)
327 packet[TCP].sport, self.tcp_port_in)
328 self.tcp_port_out = packet[TCP].sport
329 self.assert_packet_checksums_valid(packet)
330 elif packet.haslayer(UDP):
334 packet[UDP].sport, self.udp_port_in)
337 packet[UDP].sport, self.udp_port_in)
338 self.udp_port_out = packet[UDP].sport
343 packet[ICMP46].id, self.icmp_id_in)
346 packet[ICMP46].id, self.icmp_id_in)
347 self.icmp_id_out = packet[ICMP46].id
348 self.assert_packet_checksums_valid(packet)
350 self.logger.error(ppp("Unexpected or invalid packet "
351 "(outside network):", packet))
354 def verify_capture_out_ip6(self, capture, nat_ip, same_port=False,
357 Verify captured packets on outside network
359 :param capture: Captured packets
360 :param nat_ip: Translated IP address
361 :param same_port: Source port number is not translated (Default False)
362 :param dst_ip: Destination IP address (Default do not verify)
364 return self.verify_capture_out(capture, nat_ip, same_port, dst_ip,
367 def verify_capture_in(self, capture, in_if):
369 Verify captured packets on inside network
371 :param capture: Captured packets
372 :param in_if: Inside interface
374 for packet in capture:
376 self.assert_packet_checksums_valid(packet)
377 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
378 if packet.haslayer(TCP):
379 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
380 elif packet.haslayer(UDP):
381 self.assertEqual(packet[UDP].dport, self.udp_port_in)
383 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
385 self.logger.error(ppp("Unexpected or invalid packet "
386 "(inside network):", packet))
389 def verify_capture_no_translation(self, capture, ingress_if, egress_if):
391 Verify captured packet that don't have to be translated
393 :param capture: Captured packets
394 :param ingress_if: Ingress interface
395 :param egress_if: Egress interface
397 for packet in capture:
399 self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
400 self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
401 if packet.haslayer(TCP):
402 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
403 elif packet.haslayer(UDP):
404 self.assertEqual(packet[UDP].sport, self.udp_port_in)
406 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
408 self.logger.error(ppp("Unexpected or invalid packet "
409 "(inside network):", packet))
412 def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
415 Verify captured packets with ICMP errors on outside network
417 :param capture: Captured packets
418 :param src_ip: Translated IP address or IP address of VPP
419 (Default use global NAT address)
420 :param icmp_type: Type of error ICMP packet
421 we are expecting (Default 11)
424 src_ip = self.nat_addr
425 for packet in capture:
427 self.assertEqual(packet[IP].src, src_ip)
428 self.assertEqual(packet.haslayer(ICMP), 1)
430 self.assertEqual(icmp.type, icmp_type)
431 self.assertTrue(icmp.haslayer(IPerror))
432 inner_ip = icmp[IPerror]
433 if inner_ip.haslayer(TCPerror):
434 self.assertEqual(inner_ip[TCPerror].dport,
436 elif inner_ip.haslayer(UDPerror):
437 self.assertEqual(inner_ip[UDPerror].dport,
440 self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
442 self.logger.error(ppp("Unexpected or invalid packet "
443 "(outside network):", packet))
446 def verify_capture_in_with_icmp_errors(self, capture, in_if, icmp_type=11):
448 Verify captured packets with ICMP errors on inside network
450 :param capture: Captured packets
451 :param in_if: Inside interface
452 :param icmp_type: Type of error ICMP packet
453 we are expecting (Default 11)
455 for packet in capture:
457 self.assertEqual(packet[IP].dst, in_if.remote_ip4)
458 self.assertEqual(packet.haslayer(ICMP), 1)
460 self.assertEqual(icmp.type, icmp_type)
461 self.assertTrue(icmp.haslayer(IPerror))
462 inner_ip = icmp[IPerror]
463 if inner_ip.haslayer(TCPerror):
464 self.assertEqual(inner_ip[TCPerror].sport,
466 elif inner_ip.haslayer(UDPerror):
467 self.assertEqual(inner_ip[UDPerror].sport,
470 self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
472 self.logger.error(ppp("Unexpected or invalid packet "
473 "(inside network):", packet))
476 def create_stream_frag(self, src_if, dst, sport, dport, data,
477 proto=IP_PROTOS.tcp, echo_reply=False):
479 Create fragmented packet stream
481 :param src_if: Source interface
482 :param dst: Destination IPv4 address
483 :param sport: Source port
484 :param dport: Destination port
485 :param data: Payload data
486 :param proto: protocol (TCP, UDP, ICMP)
487 :param echo_reply: use echo_reply if protocol is ICMP
490 if proto == IP_PROTOS.tcp:
491 p = (IP(src=src_if.remote_ip4, dst=dst) /
492 TCP(sport=sport, dport=dport) /
494 p = p.__class__(scapy.compat.raw(p))
495 chksum = p[TCP].chksum
496 proto_header = TCP(sport=sport, dport=dport, chksum=chksum)
497 elif proto == IP_PROTOS.udp:
498 proto_header = UDP(sport=sport, dport=dport)
499 elif proto == IP_PROTOS.icmp:
501 proto_header = ICMP(id=sport, type='echo-request')
503 proto_header = ICMP(id=sport, type='echo-reply')
505 raise Exception("Unsupported protocol")
506 id = random.randint(0, 65535)
508 if proto == IP_PROTOS.tcp:
511 raw = Raw(data[0:16])
512 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
513 IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=0, id=id) /
517 if proto == IP_PROTOS.tcp:
518 raw = Raw(data[4:20])
520 raw = Raw(data[16:32])
521 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
522 IP(src=src_if.remote_ip4, dst=dst, flags="MF", frag=3, id=id,
526 if proto == IP_PROTOS.tcp:
530 p = (Ether(src=src_if.remote_mac, dst=src_if.local_mac) /
531 IP(src=src_if.remote_ip4, dst=dst, frag=5, proto=proto,
537 def reass_frags_and_verify(self, frags, src, dst):
539 Reassemble and verify fragmented packet
541 :param frags: Captured fragments
542 :param src: Source IPv4 address to verify
543 :param dst: Destination IPv4 address to verify
545 :returns: Reassembled IPv4 packet
549 self.assertEqual(p[IP].src, src)
550 self.assertEqual(p[IP].dst, dst)
551 self.assert_ip_checksum_valid(p)
552 buffer.seek(p[IP].frag * 8)
553 buffer.write(bytes(p[IP].payload))
554 ip = IP(src=frags[0][IP].src, dst=frags[0][IP].dst,
555 proto=frags[0][IP].proto)
556 if ip.proto == IP_PROTOS.tcp:
557 p = (ip / TCP(buffer.getvalue()))
558 self.logger.debug(ppp("Reassembled:", p))
559 self.assert_tcp_checksum_valid(p)
560 elif ip.proto == IP_PROTOS.udp:
561 p = (ip / UDP(buffer.getvalue()[:8]) /
562 Raw(buffer.getvalue()[8:]))
563 elif ip.proto == IP_PROTOS.icmp:
564 p = (ip / ICMP(buffer.getvalue()))
567 def verify_ipfix_nat44_ses(self, data):
569 Verify IPFIX NAT44EI session create/delete event
571 :param data: Decoded IPFIX data records
573 nat44_ses_create_num = 0
574 nat44_ses_delete_num = 0
575 self.assertEqual(6, len(data))
578 self.assertIn(scapy.compat.orb(record[230]), [4, 5])
579 if scapy.compat.orb(record[230]) == 4:
580 nat44_ses_create_num += 1
582 nat44_ses_delete_num += 1
584 self.assertEqual(self.pg0.remote_ip4,
585 str(ipaddress.IPv4Address(record[8])))
586 # postNATSourceIPv4Address
587 self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
590 self.assertEqual(struct.pack("!I", 0), record[234])
591 # protocolIdentifier/sourceTransportPort
592 # /postNAPTSourceTransportPort
593 if IP_PROTOS.icmp == scapy.compat.orb(record[4]):
594 self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
595 self.assertEqual(struct.pack("!H", self.icmp_id_out),
597 elif IP_PROTOS.tcp == scapy.compat.orb(record[4]):
598 self.assertEqual(struct.pack("!H", self.tcp_port_in),
600 self.assertEqual(struct.pack("!H", self.tcp_port_out),
602 elif IP_PROTOS.udp == scapy.compat.orb(record[4]):
603 self.assertEqual(struct.pack("!H", self.udp_port_in),
605 self.assertEqual(struct.pack("!H", self.udp_port_out),
608 self.fail("Invalid protocol")
609 self.assertEqual(3, nat44_ses_create_num)
610 self.assertEqual(3, nat44_ses_delete_num)
612 def verify_ipfix_addr_exhausted(self, data):
613 self.assertEqual(1, len(data))
616 self.assertEqual(scapy.compat.orb(record[230]), 3)
618 self.assertEqual(struct.pack("!I", 0), record[283])
620 def verify_ipfix_max_sessions(self, data, limit):
621 self.assertEqual(1, len(data))
624 self.assertEqual(scapy.compat.orb(record[230]), 13)
625 # natQuotaExceededEvent
626 self.assertEqual(struct.pack("!I", 1), record[466])
628 self.assertEqual(struct.pack("!I", limit), record[471])
630 def verify_no_nat44_user(self):
631 """ Verify that there is no NAT44EI user """
632 users = self.vapi.nat44_ei_user_dump()
633 self.assertEqual(len(users), 0)
634 users = self.statistics['/nat44-ei/total-users']
635 self.assertEqual(users[0][0], 0)
636 sessions = self.statistics['/nat44-ei/total-sessions']
637 self.assertEqual(sessions[0][0], 0)
639 def verify_syslog_apmap(self, data, is_add=True):
640 message = data.decode('utf-8')
642 message = SyslogMessage.parse(message)
643 except ParseError as e:
647 self.assertEqual(message.severity, SyslogSeverity.info)
648 self.assertEqual(message.appname, 'NAT')
649 self.assertEqual(message.msgid, 'APMADD' if is_add else 'APMDEL')
650 sd_params = message.sd.get('napmap')
651 self.assertTrue(sd_params is not None)
652 self.assertEqual(sd_params.get('IATYP'), 'IPv4')
653 self.assertEqual(sd_params.get('ISADDR'), self.pg0.remote_ip4)
654 self.assertEqual(sd_params.get('ISPORT'), "%d" % self.tcp_port_in)
655 self.assertEqual(sd_params.get('XATYP'), 'IPv4')
656 self.assertEqual(sd_params.get('XSADDR'), self.nat_addr)
657 self.assertEqual(sd_params.get('XSPORT'), "%d" % self.tcp_port_out)
658 self.assertEqual(sd_params.get('PROTO'), "%d" % IP_PROTOS.tcp)
659 self.assertTrue(sd_params.get('SSUBIX') is not None)
660 self.assertEqual(sd_params.get('SVLAN'), '0')
662 def verify_mss_value(self, pkt, mss):
663 if not pkt.haslayer(IP) or not pkt.haslayer(TCP):
664 raise TypeError("Not a TCP/IP packet")
666 for option in pkt[TCP].options:
667 if option[0] == 'MSS':
668 self.assertEqual(option[1], mss)
669 self.assert_tcp_checksum_valid(pkt)
672 def proto2layer(proto):
673 if proto == IP_PROTOS.tcp:
675 elif proto == IP_PROTOS.udp:
677 elif proto == IP_PROTOS.icmp:
680 raise Exception("Unsupported protocol")
682 def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
684 layer = self.proto2layer(proto)
686 if proto == IP_PROTOS.tcp:
687 data = b"A" * 4 + b"B" * 16 + b"C" * 3
689 data = b"A" * 16 + b"B" * 16 + b"C" * 3
690 self.port_in = random.randint(1025, 65535)
693 pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
694 self.port_in, 20, data, proto)
695 self.pg0.add_stream(pkts)
696 self.pg_enable_capture(self.pg_interfaces)
698 frags = self.pg1.get_capture(len(pkts))
699 if not dont_translate:
700 p = self.reass_frags_and_verify(frags,
704 p = self.reass_frags_and_verify(frags,
707 if proto != IP_PROTOS.icmp:
708 if not dont_translate:
709 self.assertEqual(p[layer].dport, 20)
711 self.assertNotEqual(p[layer].sport, self.port_in)
713 self.assertEqual(p[layer].sport, self.port_in)
716 if not dont_translate:
717 self.assertNotEqual(p[layer].id, self.port_in)
719 self.assertEqual(p[layer].id, self.port_in)
720 self.assertEqual(data, p[Raw].load)
723 if not dont_translate:
724 dst_addr = self.nat_addr
726 dst_addr = self.pg0.remote_ip4
727 if proto != IP_PROTOS.icmp:
729 dport = p[layer].sport
733 pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport, data,
734 proto, echo_reply=True)
735 self.pg1.add_stream(pkts)
736 self.pg_enable_capture(self.pg_interfaces)
738 frags = self.pg0.get_capture(len(pkts))
739 p = self.reass_frags_and_verify(frags,
742 if proto != IP_PROTOS.icmp:
743 self.assertEqual(p[layer].sport, 20)
744 self.assertEqual(p[layer].dport, self.port_in)
746 self.assertEqual(p[layer].id, self.port_in)
747 self.assertEqual(data, p[Raw].load)
749 def reass_hairpinning(self, server_addr, server_in_port, server_out_port,
750 host_in_port, proto=IP_PROTOS.tcp,
753 layer = self.proto2layer(proto)
755 if proto == IP_PROTOS.tcp:
756 data = b"A" * 4 + b"B" * 16 + b"C" * 3
758 data = b"A" * 16 + b"B" * 16 + b"C" * 3
760 # send packet from host to server
761 pkts = self.create_stream_frag(self.pg0,
767 self.pg0.add_stream(pkts)
768 self.pg_enable_capture(self.pg_interfaces)
770 frags = self.pg0.get_capture(len(pkts))
771 p = self.reass_frags_and_verify(frags,
774 if proto != IP_PROTOS.icmp:
776 self.assertNotEqual(p[layer].sport, host_in_port)
777 self.assertEqual(p[layer].dport, server_in_port)
780 self.assertNotEqual(p[layer].id, host_in_port)
781 self.assertEqual(data, p[Raw].load)
783 def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
785 layer = self.proto2layer(proto)
787 if proto == IP_PROTOS.tcp:
788 data = b"A" * 4 + b"B" * 16 + b"C" * 3
790 data = b"A" * 16 + b"B" * 16 + b"C" * 3
791 self.port_in = random.randint(1025, 65535)
795 pkts = self.create_stream_frag(self.pg0, self.pg1.remote_ip4,
796 self.port_in, 20, data, proto)
798 self.pg0.add_stream(pkts)
799 self.pg_enable_capture(self.pg_interfaces)
801 frags = self.pg1.get_capture(len(pkts))
802 if not dont_translate:
803 p = self.reass_frags_and_verify(frags,
807 p = self.reass_frags_and_verify(frags,
810 if proto != IP_PROTOS.icmp:
811 if not dont_translate:
812 self.assertEqual(p[layer].dport, 20)
814 self.assertNotEqual(p[layer].sport, self.port_in)
816 self.assertEqual(p[layer].sport, self.port_in)
819 if not dont_translate:
820 self.assertNotEqual(p[layer].id, self.port_in)
822 self.assertEqual(p[layer].id, self.port_in)
823 self.assertEqual(data, p[Raw].load)
826 if not dont_translate:
827 dst_addr = self.nat_addr
829 dst_addr = self.pg0.remote_ip4
830 if proto != IP_PROTOS.icmp:
832 dport = p[layer].sport
836 pkts = self.create_stream_frag(self.pg1, dst_addr, sport, dport,
837 data, proto, echo_reply=True)
839 self.pg1.add_stream(pkts)
840 self.pg_enable_capture(self.pg_interfaces)
842 frags = self.pg0.get_capture(len(pkts))
843 p = self.reass_frags_and_verify(frags,
846 if proto != IP_PROTOS.icmp:
847 self.assertEqual(p[layer].sport, 20)
848 self.assertEqual(p[layer].dport, self.port_in)
850 self.assertEqual(p[layer].id, self.port_in)
851 self.assertEqual(data, p[Raw].load)
854 def get_nat44_ei_in2out_worker_index(ip, vpp_worker_count):
855 if 0 == vpp_worker_count:
857 numeric = socket.inet_aton(ip)
858 numeric = struct.unpack("!L", numeric)[0]
859 numeric = socket.htonl(numeric)
860 h = numeric + (numeric >> 8) + (numeric >> 16) + (numeric >> 24)
861 return 1 + h % vpp_worker_count
864 class TestNAT44EI(MethodHolder):
865 """ NAT44EI Test Cases """
867 max_translations = 10240
872 super(TestNAT44EI, cls).setUpClass()
873 cls.vapi.cli("set log class nat44-ei level debug")
875 cls.tcp_port_in = 6303
876 cls.tcp_port_out = 6303
877 cls.udp_port_in = 6304
878 cls.udp_port_out = 6304
879 cls.icmp_id_in = 6305
880 cls.icmp_id_out = 6305
881 cls.nat_addr = '10.0.0.3'
882 cls.ipfix_src_port = 4739
883 cls.ipfix_domain_id = 1
884 cls.tcp_external_port = 80
885 cls.udp_external_port = 69
887 cls.create_pg_interfaces(range(10))
888 cls.interfaces = list(cls.pg_interfaces[0:4])
890 for i in cls.interfaces:
895 cls.pg0.generate_remote_hosts(3)
896 cls.pg0.configure_ipv4_neighbors()
898 cls.pg1.generate_remote_hosts(1)
899 cls.pg1.configure_ipv4_neighbors()
901 cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
902 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
903 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
905 cls.pg4._local_ip4 = "172.16.255.1"
906 cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
907 cls.pg4.set_table_ip4(10)
908 cls.pg5._local_ip4 = "172.17.255.3"
909 cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
910 cls.pg5.set_table_ip4(10)
911 cls.pg6._local_ip4 = "172.16.255.1"
912 cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
913 cls.pg6.set_table_ip4(20)
914 for i in cls.overlapping_interfaces:
922 cls.pg9.generate_remote_hosts(2)
924 cls.vapi.sw_interface_add_del_address(
925 sw_if_index=cls.pg9.sw_if_index,
926 prefix="10.0.0.1/24")
929 cls.pg9.resolve_arp()
930 cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
931 cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
932 cls.pg9.resolve_arp()
934 def plugin_enable(self):
935 self.vapi.nat44_ei_plugin_enable_disable(
936 sessions=self.max_translations,
937 users=self.max_users, enable=1)
940 super(TestNAT44EI, self).setUp()
944 super(TestNAT44EI, self).tearDown()
945 if not self.vpp_dead:
946 self.vapi.nat44_ei_ipfix_enable_disable(
947 domain_id=self.ipfix_domain_id, src_port=self.ipfix_src_port,
949 self.ipfix_src_port = 4739
950 self.ipfix_domain_id = 1
952 self.vapi.nat44_ei_plugin_enable_disable(enable=0)
953 self.vapi.cli("clear logging")
955 def test_clear_sessions(self):
956 """ NAT44EI session clearing test """
958 self.nat44_add_address(self.nat_addr)
959 flags = self.config_flags.NAT44_EI_IF_INSIDE
960 self.vapi.nat44_ei_interface_add_del_feature(
961 sw_if_index=self.pg0.sw_if_index,
962 flags=flags, is_add=1)
963 self.vapi.nat44_ei_interface_add_del_feature(
964 sw_if_index=self.pg1.sw_if_index,
967 pkts = self.create_stream_in(self.pg0, self.pg1)
968 self.pg0.add_stream(pkts)
969 self.pg_enable_capture(self.pg_interfaces)
971 capture = self.pg1.get_capture(len(pkts))
972 self.verify_capture_out(capture)
974 sessions = self.statistics['/nat44-ei/total-sessions']
975 self.assertGreater(sessions[:, 0].sum(), 0, "Session count invalid")
976 self.logger.info("sessions before clearing: %s" % sessions[0][0])
978 self.vapi.cli("clear nat44 ei sessions")
980 sessions = self.statistics['/nat44-ei/total-sessions']
981 self.assertEqual(sessions[:, 0].sum(), 0, "Session count invalid")
982 self.logger.info("sessions after clearing: %s" % sessions[0][0])
984 def test_dynamic(self):
985 """ NAT44EI dynamic translation test """
986 self.nat44_add_address(self.nat_addr)
987 flags = self.config_flags.NAT44_EI_IF_INSIDE
988 self.vapi.nat44_ei_interface_add_del_feature(
989 sw_if_index=self.pg0.sw_if_index,
990 flags=flags, is_add=1)
991 self.vapi.nat44_ei_interface_add_del_feature(
992 sw_if_index=self.pg1.sw_if_index,
996 tcpn = self.statistics['/nat44-ei/in2out/slowpath/tcp']
997 udpn = self.statistics['/nat44-ei/in2out/slowpath/udp']
998 icmpn = self.statistics['/nat44-ei/in2out/slowpath/icmp']
999 drops = self.statistics['/nat44-ei/in2out/slowpath/drops']
1001 pkts = self.create_stream_in(self.pg0, self.pg1)
1002 self.pg0.add_stream(pkts)
1003 self.pg_enable_capture(self.pg_interfaces)
1005 capture = self.pg1.get_capture(len(pkts))
1006 self.verify_capture_out(capture)
1008 if_idx = self.pg0.sw_if_index
1009 cnt = self.statistics['/nat44-ei/in2out/slowpath/tcp']
1010 self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
1011 cnt = self.statistics['/nat44-ei/in2out/slowpath/udp']
1012 self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
1013 cnt = self.statistics['/nat44-ei/in2out/slowpath/icmp']
1014 self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
1015 cnt = self.statistics['/nat44-ei/in2out/slowpath/drops']
1016 self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
1019 tcpn = self.statistics['/nat44-ei/out2in/slowpath/tcp']
1020 udpn = self.statistics['/nat44-ei/out2in/slowpath/udp']
1021 icmpn = self.statistics['/nat44-ei/out2in/slowpath/icmp']
1022 drops = self.statistics['/nat44-ei/out2in/slowpath/drops']
1024 pkts = self.create_stream_out(self.pg1)
1025 self.pg1.add_stream(pkts)
1026 self.pg_enable_capture(self.pg_interfaces)
1028 capture = self.pg0.get_capture(len(pkts))
1029 self.verify_capture_in(capture, self.pg0)
1031 if_idx = self.pg1.sw_if_index
1032 cnt = self.statistics['/nat44-ei/out2in/slowpath/tcp']
1033 self.assertEqual(cnt[:, if_idx].sum() - tcpn[:, if_idx].sum(), 2)
1034 cnt = self.statistics['/nat44-ei/out2in/slowpath/udp']
1035 self.assertEqual(cnt[:, if_idx].sum() - udpn[:, if_idx].sum(), 1)
1036 cnt = self.statistics['/nat44-ei/out2in/slowpath/icmp']
1037 self.assertEqual(cnt[:, if_idx].sum() - icmpn[:, if_idx].sum(), 1)
1038 cnt = self.statistics['/nat44-ei/out2in/slowpath/drops']
1039 self.assertEqual(cnt[:, if_idx].sum() - drops[:, if_idx].sum(), 0)
1041 users = self.statistics['/nat44-ei/total-users']
1042 self.assertEqual(users[:, 0].sum(), 1)
1043 sessions = self.statistics['/nat44-ei/total-sessions']
1044 self.assertEqual(sessions[:, 0].sum(), 3)
1046 def test_dynamic_icmp_errors_in2out_ttl_1(self):
1047 """ NAT44EI handling of client packets with TTL=1 """
1049 self.nat44_add_address(self.nat_addr)
1050 flags = self.config_flags.NAT44_EI_IF_INSIDE
1051 self.vapi.nat44_ei_interface_add_del_feature(
1052 sw_if_index=self.pg0.sw_if_index,
1053 flags=flags, is_add=1)
1054 self.vapi.nat44_ei_interface_add_del_feature(
1055 sw_if_index=self.pg1.sw_if_index,
1058 # Client side - generate traffic
1059 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
1060 self.pg0.add_stream(pkts)
1061 self.pg_enable_capture(self.pg_interfaces)
1064 # Client side - verify ICMP type 11 packets
1065 capture = self.pg0.get_capture(len(pkts))
1066 self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1068 def test_dynamic_icmp_errors_out2in_ttl_1(self):
1069 """ NAT44EI handling of server packets with TTL=1 """
1071 self.nat44_add_address(self.nat_addr)
1072 flags = self.config_flags.NAT44_EI_IF_INSIDE
1073 self.vapi.nat44_ei_interface_add_del_feature(
1074 sw_if_index=self.pg0.sw_if_index,
1075 flags=flags, is_add=1)
1076 self.vapi.nat44_ei_interface_add_del_feature(
1077 sw_if_index=self.pg1.sw_if_index,
1080 # Client side - create sessions
1081 pkts = self.create_stream_in(self.pg0, self.pg1)
1082 self.pg0.add_stream(pkts)
1083 self.pg_enable_capture(self.pg_interfaces)
1086 # Server side - generate traffic
1087 capture = self.pg1.get_capture(len(pkts))
1088 self.verify_capture_out(capture)
1089 pkts = self.create_stream_out(self.pg1, ttl=1)
1090 self.pg1.add_stream(pkts)
1091 self.pg_enable_capture(self.pg_interfaces)
1094 # Server side - verify ICMP type 11 packets
1095 capture = self.pg1.get_capture(len(pkts))
1096 self.verify_capture_out_with_icmp_errors(capture,
1097 src_ip=self.pg1.local_ip4)
1099 def test_dynamic_icmp_errors_in2out_ttl_2(self):
1100 """ NAT44EI handling of error responses to client packets with TTL=2
1103 self.nat44_add_address(self.nat_addr)
1104 flags = self.config_flags.NAT44_EI_IF_INSIDE
1105 self.vapi.nat44_ei_interface_add_del_feature(
1106 sw_if_index=self.pg0.sw_if_index,
1107 flags=flags, is_add=1)
1108 self.vapi.nat44_ei_interface_add_del_feature(
1109 sw_if_index=self.pg1.sw_if_index,
1112 # Client side - generate traffic
1113 pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
1114 self.pg0.add_stream(pkts)
1115 self.pg_enable_capture(self.pg_interfaces)
1118 # Server side - simulate ICMP type 11 response
1119 capture = self.pg1.get_capture(len(pkts))
1120 pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1121 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
1122 ICMP(type=11) / packet[IP] for packet in capture]
1123 self.pg1.add_stream(pkts)
1124 self.pg_enable_capture(self.pg_interfaces)
1127 # Client side - verify ICMP type 11 packets
1128 capture = self.pg0.get_capture(len(pkts))
1129 self.verify_capture_in_with_icmp_errors(capture, self.pg0)
1131 def test_dynamic_icmp_errors_out2in_ttl_2(self):
1132 """ NAT44EI handling of error responses to server packets with TTL=2
1135 self.nat44_add_address(self.nat_addr)
1136 flags = self.config_flags.NAT44_EI_IF_INSIDE
1137 self.vapi.nat44_ei_interface_add_del_feature(
1138 sw_if_index=self.pg0.sw_if_index,
1139 flags=flags, is_add=1)
1140 self.vapi.nat44_ei_interface_add_del_feature(
1141 sw_if_index=self.pg1.sw_if_index,
1144 # Client side - create sessions
1145 pkts = self.create_stream_in(self.pg0, self.pg1)
1146 self.pg0.add_stream(pkts)
1147 self.pg_enable_capture(self.pg_interfaces)
1150 # Server side - generate traffic
1151 capture = self.pg1.get_capture(len(pkts))
1152 self.verify_capture_out(capture)
1153 pkts = self.create_stream_out(self.pg1, ttl=2)
1154 self.pg1.add_stream(pkts)
1155 self.pg_enable_capture(self.pg_interfaces)
1158 # Client side - simulate ICMP type 11 response
1159 capture = self.pg0.get_capture(len(pkts))
1160 pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1161 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
1162 ICMP(type=11) / packet[IP] for packet in capture]
1163 self.pg0.add_stream(pkts)
1164 self.pg_enable_capture(self.pg_interfaces)
1167 # Server side - verify ICMP type 11 packets
1168 capture = self.pg1.get_capture(len(pkts))
1169 self.verify_capture_out_with_icmp_errors(capture)
1171 def test_ping_out_interface_from_outside(self):
1172 """ NAT44EI ping out interface from outside network """
1174 self.nat44_add_address(self.nat_addr)
1175 flags = self.config_flags.NAT44_EI_IF_INSIDE
1176 self.vapi.nat44_ei_interface_add_del_feature(
1177 sw_if_index=self.pg0.sw_if_index,
1178 flags=flags, is_add=1)
1179 self.vapi.nat44_ei_interface_add_del_feature(
1180 sw_if_index=self.pg1.sw_if_index,
1183 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1184 IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
1185 ICMP(id=self.icmp_id_out, type='echo-request'))
1187 self.pg1.add_stream(pkts)
1188 self.pg_enable_capture(self.pg_interfaces)
1190 capture = self.pg1.get_capture(len(pkts))
1193 self.assertEqual(packet[IP].src, self.pg1.local_ip4)
1194 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
1195 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
1196 self.assertEqual(packet[ICMP].type, 0) # echo reply
1198 self.logger.error(ppp("Unexpected or invalid packet "
1199 "(outside network):", packet))
1202 def test_ping_internal_host_from_outside(self):
1203 """ NAT44EI ping internal host from outside network """
1205 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr)
1206 flags = self.config_flags.NAT44_EI_IF_INSIDE
1207 self.vapi.nat44_ei_interface_add_del_feature(
1208 sw_if_index=self.pg0.sw_if_index,
1209 flags=flags, is_add=1)
1210 self.vapi.nat44_ei_interface_add_del_feature(
1211 sw_if_index=self.pg1.sw_if_index,
1215 pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
1216 IP(src=self.pg1.remote_ip4, dst=self.nat_addr, ttl=64) /
1217 ICMP(id=self.icmp_id_out, type='echo-request'))
1218 self.pg1.add_stream(pkt)
1219 self.pg_enable_capture(self.pg_interfaces)
1221 capture = self.pg0.get_capture(1)
1222 self.verify_capture_in(capture, self.pg0)
1223 self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1226 pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1227 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
1228 ICMP(id=self.icmp_id_in, type='echo-reply'))
1229 self.pg0.add_stream(pkt)
1230 self.pg_enable_capture(self.pg_interfaces)
1232 capture = self.pg1.get_capture(1)
1233 self.verify_capture_out(capture, same_port=True)
1234 self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
1236 def test_forwarding(self):
1237 """ NAT44EI forwarding test """
1239 flags = self.config_flags.NAT44_EI_IF_INSIDE
1240 self.vapi.nat44_ei_interface_add_del_feature(
1241 sw_if_index=self.pg0.sw_if_index,
1242 flags=flags, is_add=1)
1243 self.vapi.nat44_ei_interface_add_del_feature(
1244 sw_if_index=self.pg1.sw_if_index,
1246 self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
1248 real_ip = self.pg0.remote_ip4
1249 alias_ip = self.nat_addr
1250 flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1251 self.vapi.nat44_ei_add_del_static_mapping(
1252 is_add=1, local_ip_address=real_ip,
1253 external_ip_address=alias_ip,
1254 external_sw_if_index=0xFFFFFFFF,
1258 # static mapping match
1260 pkts = self.create_stream_out(self.pg1)
1261 self.pg1.add_stream(pkts)
1262 self.pg_enable_capture(self.pg_interfaces)
1264 capture = self.pg0.get_capture(len(pkts))
1265 self.verify_capture_in(capture, self.pg0)
1267 pkts = self.create_stream_in(self.pg0, self.pg1)
1268 self.pg0.add_stream(pkts)
1269 self.pg_enable_capture(self.pg_interfaces)
1271 capture = self.pg1.get_capture(len(pkts))
1272 self.verify_capture_out(capture, same_port=True)
1274 # no static mapping match
1276 host0 = self.pg0.remote_hosts[0]
1277 self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
1279 pkts = self.create_stream_out(self.pg1,
1280 dst_ip=self.pg0.remote_ip4,
1281 use_inside_ports=True)
1282 self.pg1.add_stream(pkts)
1283 self.pg_enable_capture(self.pg_interfaces)
1285 capture = self.pg0.get_capture(len(pkts))
1286 self.verify_capture_in(capture, self.pg0)
1288 pkts = self.create_stream_in(self.pg0, self.pg1)
1289 self.pg0.add_stream(pkts)
1290 self.pg_enable_capture(self.pg_interfaces)
1292 capture = self.pg1.get_capture(len(pkts))
1293 self.verify_capture_out(capture, nat_ip=self.pg0.remote_ip4,
1296 self.pg0.remote_hosts[0] = host0
1299 self.vapi.nat44_ei_forwarding_enable_disable(enable=0)
1300 flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1301 self.vapi.nat44_ei_add_del_static_mapping(
1303 local_ip_address=real_ip,
1304 external_ip_address=alias_ip,
1305 external_sw_if_index=0xFFFFFFFF,
1308 def test_static_in(self):
1309 """ NAT44EI 1:1 NAT initialized from inside network """
1311 nat_ip = "10.0.0.10"
1312 self.tcp_port_out = 6303
1313 self.udp_port_out = 6304
1314 self.icmp_id_out = 6305
1316 self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1317 flags = self.config_flags.NAT44_EI_IF_INSIDE
1318 self.vapi.nat44_ei_interface_add_del_feature(
1319 sw_if_index=self.pg0.sw_if_index,
1320 flags=flags, is_add=1)
1321 self.vapi.nat44_ei_interface_add_del_feature(
1322 sw_if_index=self.pg1.sw_if_index,
1324 sm = self.vapi.nat44_ei_static_mapping_dump()
1325 self.assertEqual(len(sm), 1)
1326 self.assertEqual(sm[0].tag, '')
1327 self.assertEqual(sm[0].protocol, 0)
1328 self.assertEqual(sm[0].local_port, 0)
1329 self.assertEqual(sm[0].external_port, 0)
1332 pkts = self.create_stream_in(self.pg0, self.pg1)
1333 self.pg0.add_stream(pkts)
1334 self.pg_enable_capture(self.pg_interfaces)
1336 capture = self.pg1.get_capture(len(pkts))
1337 self.verify_capture_out(capture, nat_ip, True)
1340 pkts = self.create_stream_out(self.pg1, nat_ip)
1341 self.pg1.add_stream(pkts)
1342 self.pg_enable_capture(self.pg_interfaces)
1344 capture = self.pg0.get_capture(len(pkts))
1345 self.verify_capture_in(capture, self.pg0)
1347 def test_static_out(self):
1348 """ NAT44EI 1:1 NAT initialized from outside network """
1350 nat_ip = "10.0.0.20"
1351 self.tcp_port_out = 6303
1352 self.udp_port_out = 6304
1353 self.icmp_id_out = 6305
1356 self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip, tag=tag)
1357 flags = self.config_flags.NAT44_EI_IF_INSIDE
1358 self.vapi.nat44_ei_interface_add_del_feature(
1359 sw_if_index=self.pg0.sw_if_index,
1360 flags=flags, is_add=1)
1361 self.vapi.nat44_ei_interface_add_del_feature(
1362 sw_if_index=self.pg1.sw_if_index,
1364 sm = self.vapi.nat44_ei_static_mapping_dump()
1365 self.assertEqual(len(sm), 1)
1366 self.assertEqual(sm[0].tag, tag)
1369 pkts = self.create_stream_out(self.pg1, nat_ip)
1370 self.pg1.add_stream(pkts)
1371 self.pg_enable_capture(self.pg_interfaces)
1373 capture = self.pg0.get_capture(len(pkts))
1374 self.verify_capture_in(capture, self.pg0)
1377 pkts = self.create_stream_in(self.pg0, self.pg1)
1378 self.pg0.add_stream(pkts)
1379 self.pg_enable_capture(self.pg_interfaces)
1381 capture = self.pg1.get_capture(len(pkts))
1382 self.verify_capture_out(capture, nat_ip, True)
1384 def test_static_with_port_in(self):
1385 """ NAT44EI 1:1 NAPT initialized from inside network """
1387 self.tcp_port_out = 3606
1388 self.udp_port_out = 3607
1389 self.icmp_id_out = 3608
1391 self.nat44_add_address(self.nat_addr)
1392 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1393 self.tcp_port_in, self.tcp_port_out,
1394 proto=IP_PROTOS.tcp)
1395 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1396 self.udp_port_in, self.udp_port_out,
1397 proto=IP_PROTOS.udp)
1398 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1399 self.icmp_id_in, self.icmp_id_out,
1400 proto=IP_PROTOS.icmp)
1401 flags = self.config_flags.NAT44_EI_IF_INSIDE
1402 self.vapi.nat44_ei_interface_add_del_feature(
1403 sw_if_index=self.pg0.sw_if_index,
1404 flags=flags, is_add=1)
1405 self.vapi.nat44_ei_interface_add_del_feature(
1406 sw_if_index=self.pg1.sw_if_index,
1410 pkts = self.create_stream_in(self.pg0, self.pg1)
1411 self.pg0.add_stream(pkts)
1412 self.pg_enable_capture(self.pg_interfaces)
1414 capture = self.pg1.get_capture(len(pkts))
1415 self.verify_capture_out(capture)
1418 pkts = self.create_stream_out(self.pg1)
1419 self.pg1.add_stream(pkts)
1420 self.pg_enable_capture(self.pg_interfaces)
1422 capture = self.pg0.get_capture(len(pkts))
1423 self.verify_capture_in(capture, self.pg0)
1425 def test_static_with_port_out(self):
1426 """ NAT44EI 1:1 NAPT initialized from outside network """
1428 self.tcp_port_out = 30606
1429 self.udp_port_out = 30607
1430 self.icmp_id_out = 30608
1432 self.nat44_add_address(self.nat_addr)
1433 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1434 self.tcp_port_in, self.tcp_port_out,
1435 proto=IP_PROTOS.tcp)
1436 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1437 self.udp_port_in, self.udp_port_out,
1438 proto=IP_PROTOS.udp)
1439 self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
1440 self.icmp_id_in, self.icmp_id_out,
1441 proto=IP_PROTOS.icmp)
1442 flags = self.config_flags.NAT44_EI_IF_INSIDE
1443 self.vapi.nat44_ei_interface_add_del_feature(
1444 sw_if_index=self.pg0.sw_if_index,
1445 flags=flags, is_add=1)
1446 self.vapi.nat44_ei_interface_add_del_feature(
1447 sw_if_index=self.pg1.sw_if_index,
1451 pkts = self.create_stream_out(self.pg1)
1452 self.pg1.add_stream(pkts)
1453 self.pg_enable_capture(self.pg_interfaces)
1455 capture = self.pg0.get_capture(len(pkts))
1456 self.verify_capture_in(capture, self.pg0)
1459 pkts = self.create_stream_in(self.pg0, self.pg1)
1460 self.pg0.add_stream(pkts)
1461 self.pg_enable_capture(self.pg_interfaces)
1463 capture = self.pg1.get_capture(len(pkts))
1464 self.verify_capture_out(capture)
1466 def test_static_vrf_aware(self):
1467 """ NAT44EI 1:1 NAT VRF awareness """
1469 nat_ip1 = "10.0.0.30"
1470 nat_ip2 = "10.0.0.40"
1471 self.tcp_port_out = 6303
1472 self.udp_port_out = 6304
1473 self.icmp_id_out = 6305
1475 self.nat44_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
1477 self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
1479 flags = self.config_flags.NAT44_EI_IF_INSIDE
1480 self.vapi.nat44_ei_interface_add_del_feature(
1481 sw_if_index=self.pg3.sw_if_index,
1483 self.vapi.nat44_ei_interface_add_del_feature(
1484 sw_if_index=self.pg0.sw_if_index,
1485 flags=flags, is_add=1)
1486 self.vapi.nat44_ei_interface_add_del_feature(
1487 sw_if_index=self.pg4.sw_if_index,
1488 flags=flags, is_add=1)
1490 # inside interface VRF match NAT44EI static mapping VRF
1491 pkts = self.create_stream_in(self.pg4, self.pg3)
1492 self.pg4.add_stream(pkts)
1493 self.pg_enable_capture(self.pg_interfaces)
1495 capture = self.pg3.get_capture(len(pkts))
1496 self.verify_capture_out(capture, nat_ip1, True)
1498 # inside interface VRF don't match NAT44EI static mapping VRF (packets
1500 pkts = self.create_stream_in(self.pg0, self.pg3)
1501 self.pg0.add_stream(pkts)
1502 self.pg_enable_capture(self.pg_interfaces)
1504 self.pg3.assert_nothing_captured()
1506 def test_dynamic_to_static(self):
1507 """ NAT44EI Switch from dynamic translation to 1:1NAT """
1508 nat_ip = "10.0.0.10"
1509 self.tcp_port_out = 6303
1510 self.udp_port_out = 6304
1511 self.icmp_id_out = 6305
1513 self.nat44_add_address(self.nat_addr)
1514 flags = self.config_flags.NAT44_EI_IF_INSIDE
1515 self.vapi.nat44_ei_interface_add_del_feature(
1516 sw_if_index=self.pg0.sw_if_index,
1517 flags=flags, is_add=1)
1518 self.vapi.nat44_ei_interface_add_del_feature(
1519 sw_if_index=self.pg1.sw_if_index,
1523 pkts = self.create_stream_in(self.pg0, self.pg1)
1524 self.pg0.add_stream(pkts)
1525 self.pg_enable_capture(self.pg_interfaces)
1527 capture = self.pg1.get_capture(len(pkts))
1528 self.verify_capture_out(capture)
1531 self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
1532 sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
1533 self.assertEqual(len(sessions), 0)
1534 pkts = self.create_stream_in(self.pg0, self.pg1)
1535 self.pg0.add_stream(pkts)
1536 self.pg_enable_capture(self.pg_interfaces)
1538 capture = self.pg1.get_capture(len(pkts))
1539 self.verify_capture_out(capture, nat_ip, True)
1541 def test_identity_nat(self):
1542 """ NAT44EI Identity NAT """
1543 flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1544 self.vapi.nat44_ei_add_del_identity_mapping(
1545 ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
1546 flags=flags, is_add=1)
1547 flags = self.config_flags.NAT44_EI_IF_INSIDE
1548 self.vapi.nat44_ei_interface_add_del_feature(
1549 sw_if_index=self.pg0.sw_if_index,
1550 flags=flags, is_add=1)
1551 self.vapi.nat44_ei_interface_add_del_feature(
1552 sw_if_index=self.pg1.sw_if_index,
1555 p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
1556 IP(src=self.pg1.remote_ip4, dst=self.pg0.remote_ip4) /
1557 TCP(sport=12345, dport=56789))
1558 self.pg1.add_stream(p)
1559 self.pg_enable_capture(self.pg_interfaces)
1561 capture = self.pg0.get_capture(1)
1566 self.assertEqual(ip.dst, self.pg0.remote_ip4)
1567 self.assertEqual(ip.src, self.pg1.remote_ip4)
1568 self.assertEqual(tcp.dport, 56789)
1569 self.assertEqual(tcp.sport, 12345)
1570 self.assert_packet_checksums_valid(p)
1572 self.logger.error(ppp("Unexpected or invalid packet:", p))
1575 sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
1576 self.assertEqual(len(sessions), 0)
1577 flags = self.config_flags.NAT44_EI_ADDR_ONLY_MAPPING
1578 self.vapi.nat44_ei_add_del_identity_mapping(
1579 ip_address=self.pg0.remote_ip4, sw_if_index=0xFFFFFFFF,
1580 flags=flags, vrf_id=1, is_add=1)
1581 identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
1582 self.assertEqual(len(identity_mappings), 2)
1584 def test_multiple_inside_interfaces(self):
1585 """ NAT44EI multiple non-overlapping address space inside interfaces
1588 self.nat44_add_address(self.nat_addr)
1589 flags = self.config_flags.NAT44_EI_IF_INSIDE
1590 self.vapi.nat44_ei_interface_add_del_feature(
1591 sw_if_index=self.pg0.sw_if_index,
1592 flags=flags, is_add=1)
1593 self.vapi.nat44_ei_interface_add_del_feature(
1594 sw_if_index=self.pg1.sw_if_index,
1595 flags=flags, is_add=1)
1596 self.vapi.nat44_ei_interface_add_del_feature(
1597 sw_if_index=self.pg3.sw_if_index,
1600 # between two NAT44EI inside interfaces (no translation)
1601 pkts = self.create_stream_in(self.pg0, self.pg1)
1602 self.pg0.add_stream(pkts)
1603 self.pg_enable_capture(self.pg_interfaces)
1605 capture = self.pg1.get_capture(len(pkts))
1606 self.verify_capture_no_translation(capture, self.pg0, self.pg1)
1608 # from inside to interface without translation
1609 pkts = self.create_stream_in(self.pg0, self.pg2)
1610 self.pg0.add_stream(pkts)
1611 self.pg_enable_capture(self.pg_interfaces)
1613 capture = self.pg2.get_capture(len(pkts))
1614 self.verify_capture_no_translation(capture, self.pg0, self.pg2)
1616 # in2out 1st interface
1617 pkts = self.create_stream_in(self.pg0, self.pg3)
1618 self.pg0.add_stream(pkts)
1619 self.pg_enable_capture(self.pg_interfaces)
1621 capture = self.pg3.get_capture(len(pkts))
1622 self.verify_capture_out(capture)
1624 # out2in 1st interface
1625 pkts = self.create_stream_out(self.pg3)
1626 self.pg3.add_stream(pkts)
1627 self.pg_enable_capture(self.pg_interfaces)
1629 capture = self.pg0.get_capture(len(pkts))
1630 self.verify_capture_in(capture, self.pg0)
1632 # in2out 2nd interface
1633 pkts = self.create_stream_in(self.pg1, self.pg3)
1634 self.pg1.add_stream(pkts)
1635 self.pg_enable_capture(self.pg_interfaces)
1637 capture = self.pg3.get_capture(len(pkts))
1638 self.verify_capture_out(capture)
1640 # out2in 2nd interface
1641 pkts = self.create_stream_out(self.pg3)
1642 self.pg3.add_stream(pkts)
1643 self.pg_enable_capture(self.pg_interfaces)
1645 capture = self.pg1.get_capture(len(pkts))
1646 self.verify_capture_in(capture, self.pg1)
1648 def test_inside_overlapping_interfaces(self):
1649 """ NAT44EI multiple inside interfaces with overlapping address space
1652 static_nat_ip = "10.0.0.10"
1653 self.nat44_add_address(self.nat_addr)
1654 flags = self.config_flags.NAT44_EI_IF_INSIDE
1655 self.vapi.nat44_ei_interface_add_del_feature(
1656 sw_if_index=self.pg3.sw_if_index,
1658 self.vapi.nat44_ei_interface_add_del_feature(
1659 sw_if_index=self.pg4.sw_if_index,
1660 flags=flags, is_add=1)
1661 self.vapi.nat44_ei_interface_add_del_feature(
1662 sw_if_index=self.pg5.sw_if_index,
1663 flags=flags, is_add=1)
1664 self.vapi.nat44_ei_interface_add_del_feature(
1665 sw_if_index=self.pg6.sw_if_index,
1666 flags=flags, is_add=1)
1667 self.nat44_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
1670 # between NAT44EI inside interfaces with same VRF (no translation)
1671 pkts = self.create_stream_in(self.pg4, self.pg5)
1672 self.pg4.add_stream(pkts)
1673 self.pg_enable_capture(self.pg_interfaces)
1675 capture = self.pg5.get_capture(len(pkts))
1676 self.verify_capture_no_translation(capture, self.pg4, self.pg5)
1678 # between NAT44EI inside interfaces with different VRF (hairpinning)
1679 p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
1680 IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
1681 TCP(sport=1234, dport=5678))
1682 self.pg4.add_stream(p)
1683 self.pg_enable_capture(self.pg_interfaces)
1685 capture = self.pg6.get_capture(1)
1690 self.assertEqual(ip.src, self.nat_addr)
1691 self.assertEqual(ip.dst, self.pg6.remote_ip4)
1692 self.assertNotEqual(tcp.sport, 1234)
1693 self.assertEqual(tcp.dport, 5678)
1695 self.logger.error(ppp("Unexpected or invalid packet:", p))
1698 # in2out 1st interface
1699 pkts = self.create_stream_in(self.pg4, self.pg3)
1700 self.pg4.add_stream(pkts)
1701 self.pg_enable_capture(self.pg_interfaces)
1703 capture = self.pg3.get_capture(len(pkts))
1704 self.verify_capture_out(capture)
1706 # out2in 1st interface
1707 pkts = self.create_stream_out(self.pg3)
1708 self.pg3.add_stream(pkts)
1709 self.pg_enable_capture(self.pg_interfaces)
1711 capture = self.pg4.get_capture(len(pkts))
1712 self.verify_capture_in(capture, self.pg4)
1714 # in2out 2nd interface
1715 pkts = self.create_stream_in(self.pg5, self.pg3)
1716 self.pg5.add_stream(pkts)
1717 self.pg_enable_capture(self.pg_interfaces)
1719 capture = self.pg3.get_capture(len(pkts))
1720 self.verify_capture_out(capture)
1722 # out2in 2nd interface
1723 pkts = self.create_stream_out(self.pg3)
1724 self.pg3.add_stream(pkts)
1725 self.pg_enable_capture(self.pg_interfaces)
1727 capture = self.pg5.get_capture(len(pkts))
1728 self.verify_capture_in(capture, self.pg5)
1731 addresses = self.vapi.nat44_ei_address_dump()
1732 self.assertEqual(len(addresses), 1)
1733 sessions = self.vapi.nat44_ei_user_session_dump(
1734 self.pg5.remote_ip4, 10)
1735 self.assertEqual(len(sessions), 3)
1736 for session in sessions:
1737 self.assertFalse(session.flags &
1738 self.config_flags.NAT44_EI_STATIC_MAPPING)
1739 self.assertEqual(str(session.inside_ip_address),
1740 self.pg5.remote_ip4)
1741 self.assertEqual(session.outside_ip_address,
1742 addresses[0].ip_address)
1743 self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
1744 self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
1745 self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
1746 self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
1747 self.assertEqual(sessions[1].inside_port, self.udp_port_in)
1748 self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
1749 self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
1750 self.assertEqual(sessions[1].outside_port, self.udp_port_out)
1751 self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
1753 # in2out 3rd interface
1754 pkts = self.create_stream_in(self.pg6, self.pg3)
1755 self.pg6.add_stream(pkts)
1756 self.pg_enable_capture(self.pg_interfaces)
1758 capture = self.pg3.get_capture(len(pkts))
1759 self.verify_capture_out(capture, static_nat_ip, True)
1761 # out2in 3rd interface
1762 pkts = self.create_stream_out(self.pg3, static_nat_ip)
1763 self.pg3.add_stream(pkts)
1764 self.pg_enable_capture(self.pg_interfaces)
1766 capture = self.pg6.get_capture(len(pkts))
1767 self.verify_capture_in(capture, self.pg6)
1769 # general user and session dump verifications
1770 users = self.vapi.nat44_ei_user_dump()
1771 self.assertGreaterEqual(len(users), 3)
1772 addresses = self.vapi.nat44_ei_address_dump()
1773 self.assertEqual(len(addresses), 1)
1775 sessions = self.vapi.nat44_ei_user_session_dump(user.ip_address,
1777 for session in sessions:
1778 self.assertEqual(user.ip_address, session.inside_ip_address)
1779 self.assertTrue(session.total_bytes > session.total_pkts > 0)
1780 self.assertTrue(session.protocol in
1781 [IP_PROTOS.tcp, IP_PROTOS.udp,
1785 sessions = self.vapi.nat44_ei_user_session_dump(
1786 self.pg4.remote_ip4, 10)
1787 self.assertGreaterEqual(len(sessions), 4)
1788 for session in sessions:
1790 session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
1791 self.assertEqual(str(session.inside_ip_address),
1792 self.pg4.remote_ip4)
1793 self.assertEqual(session.outside_ip_address,
1794 addresses[0].ip_address)
1797 sessions = self.vapi.nat44_ei_user_session_dump(
1798 self.pg6.remote_ip4, 20)
1799 self.assertGreaterEqual(len(sessions), 3)
1800 for session in sessions:
1802 session.flags & self.config_flags.NAT44_EI_STATIC_MAPPING)
1803 self.assertEqual(str(session.inside_ip_address),
1804 self.pg6.remote_ip4)
1805 self.assertEqual(str(session.outside_ip_address),
1807 self.assertTrue(session.inside_port in
1808 [self.tcp_port_in, self.udp_port_in,
1811 def test_hairpinning(self):
1812 """ NAT44EI hairpinning - 1:1 NAPT """
1814 host = self.pg0.remote_hosts[0]
1815 server = self.pg0.remote_hosts[1]
1818 server_in_port = 5678
1819 server_out_port = 8765
1821 self.nat44_add_address(self.nat_addr)
1822 flags = self.config_flags.NAT44_EI_IF_INSIDE
1823 self.vapi.nat44_ei_interface_add_del_feature(
1824 sw_if_index=self.pg0.sw_if_index,
1825 flags=flags, is_add=1)
1826 self.vapi.nat44_ei_interface_add_del_feature(
1827 sw_if_index=self.pg1.sw_if_index,
1830 # add static mapping for server
1831 self.nat44_add_static_mapping(server.ip4, self.nat_addr,
1832 server_in_port, server_out_port,
1833 proto=IP_PROTOS.tcp)
1835 cnt = self.statistics['/nat44-ei/hairpinning']
1836 # send packet from host to server
1837 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
1838 IP(src=host.ip4, dst=self.nat_addr) /
1839 TCP(sport=host_in_port, dport=server_out_port))
1840 self.pg0.add_stream(p)
1841 self.pg_enable_capture(self.pg_interfaces)
1843 capture = self.pg0.get_capture(1)
1848 self.assertEqual(ip.src, self.nat_addr)
1849 self.assertEqual(ip.dst, server.ip4)
1850 self.assertNotEqual(tcp.sport, host_in_port)
1851 self.assertEqual(tcp.dport, server_in_port)
1852 self.assert_packet_checksums_valid(p)
1853 host_out_port = tcp.sport
1855 self.logger.error(ppp("Unexpected or invalid packet:", p))
1858 after = self.statistics['/nat44-ei/hairpinning']
1859 if_idx = self.pg0.sw_if_index
1860 self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
1862 # send reply from server to host
1863 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
1864 IP(src=server.ip4, dst=self.nat_addr) /
1865 TCP(sport=server_in_port, dport=host_out_port))
1866 self.pg0.add_stream(p)
1867 self.pg_enable_capture(self.pg_interfaces)
1869 capture = self.pg0.get_capture(1)
1874 self.assertEqual(ip.src, self.nat_addr)
1875 self.assertEqual(ip.dst, host.ip4)
1876 self.assertEqual(tcp.sport, server_out_port)
1877 self.assertEqual(tcp.dport, host_in_port)
1878 self.assert_packet_checksums_valid(p)
1880 self.logger.error(ppp("Unexpected or invalid packet:", p))
1883 after = self.statistics['/nat44-ei/hairpinning']
1884 if_idx = self.pg0.sw_if_index
1885 self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
1886 2+(1 if self.vpp_worker_count > 0 else 0))
1888 def test_hairpinning2(self):
1889 """ NAT44EI hairpinning - 1:1 NAT"""
1891 server1_nat_ip = "10.0.0.10"
1892 server2_nat_ip = "10.0.0.11"
1893 host = self.pg0.remote_hosts[0]
1894 server1 = self.pg0.remote_hosts[1]
1895 server2 = self.pg0.remote_hosts[2]
1896 server_tcp_port = 22
1897 server_udp_port = 20
1899 self.nat44_add_address(self.nat_addr)
1900 flags = self.config_flags.NAT44_EI_IF_INSIDE
1901 self.vapi.nat44_ei_interface_add_del_feature(
1902 sw_if_index=self.pg0.sw_if_index,
1903 flags=flags, is_add=1)
1904 self.vapi.nat44_ei_interface_add_del_feature(
1905 sw_if_index=self.pg1.sw_if_index,
1908 # add static mapping for servers
1909 self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
1910 self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
1914 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1915 IP(src=host.ip4, dst=server1_nat_ip) /
1916 TCP(sport=self.tcp_port_in, dport=server_tcp_port))
1918 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1919 IP(src=host.ip4, dst=server1_nat_ip) /
1920 UDP(sport=self.udp_port_in, dport=server_udp_port))
1922 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1923 IP(src=host.ip4, dst=server1_nat_ip) /
1924 ICMP(id=self.icmp_id_in, type='echo-request'))
1926 self.pg0.add_stream(pkts)
1927 self.pg_enable_capture(self.pg_interfaces)
1929 capture = self.pg0.get_capture(len(pkts))
1930 for packet in capture:
1932 self.assertEqual(packet[IP].src, self.nat_addr)
1933 self.assertEqual(packet[IP].dst, server1.ip4)
1934 if packet.haslayer(TCP):
1935 self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
1936 self.assertEqual(packet[TCP].dport, server_tcp_port)
1937 self.tcp_port_out = packet[TCP].sport
1938 self.assert_packet_checksums_valid(packet)
1939 elif packet.haslayer(UDP):
1940 self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
1941 self.assertEqual(packet[UDP].dport, server_udp_port)
1942 self.udp_port_out = packet[UDP].sport
1944 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
1945 self.icmp_id_out = packet[ICMP].id
1947 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1952 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1953 IP(src=server1.ip4, dst=self.nat_addr) /
1954 TCP(sport=server_tcp_port, dport=self.tcp_port_out))
1956 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1957 IP(src=server1.ip4, dst=self.nat_addr) /
1958 UDP(sport=server_udp_port, dport=self.udp_port_out))
1960 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1961 IP(src=server1.ip4, dst=self.nat_addr) /
1962 ICMP(id=self.icmp_id_out, type='echo-reply'))
1964 self.pg0.add_stream(pkts)
1965 self.pg_enable_capture(self.pg_interfaces)
1967 capture = self.pg0.get_capture(len(pkts))
1968 for packet in capture:
1970 self.assertEqual(packet[IP].src, server1_nat_ip)
1971 self.assertEqual(packet[IP].dst, host.ip4)
1972 if packet.haslayer(TCP):
1973 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
1974 self.assertEqual(packet[TCP].sport, server_tcp_port)
1975 self.assert_packet_checksums_valid(packet)
1976 elif packet.haslayer(UDP):
1977 self.assertEqual(packet[UDP].dport, self.udp_port_in)
1978 self.assertEqual(packet[UDP].sport, server_udp_port)
1980 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
1982 self.logger.error(ppp("Unexpected or invalid packet:", packet))
1985 # server2 to server1
1987 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1988 IP(src=server2.ip4, dst=server1_nat_ip) /
1989 TCP(sport=self.tcp_port_in, dport=server_tcp_port))
1991 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1992 IP(src=server2.ip4, dst=server1_nat_ip) /
1993 UDP(sport=self.udp_port_in, dport=server_udp_port))
1995 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
1996 IP(src=server2.ip4, dst=server1_nat_ip) /
1997 ICMP(id=self.icmp_id_in, type='echo-request'))
1999 self.pg0.add_stream(pkts)
2000 self.pg_enable_capture(self.pg_interfaces)
2002 capture = self.pg0.get_capture(len(pkts))
2003 for packet in capture:
2005 self.assertEqual(packet[IP].src, server2_nat_ip)
2006 self.assertEqual(packet[IP].dst, server1.ip4)
2007 if packet.haslayer(TCP):
2008 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
2009 self.assertEqual(packet[TCP].dport, server_tcp_port)
2010 self.tcp_port_out = packet[TCP].sport
2011 self.assert_packet_checksums_valid(packet)
2012 elif packet.haslayer(UDP):
2013 self.assertEqual(packet[UDP].sport, self.udp_port_in)
2014 self.assertEqual(packet[UDP].dport, server_udp_port)
2015 self.udp_port_out = packet[UDP].sport
2017 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2018 self.icmp_id_out = packet[ICMP].id
2020 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2023 # server1 to server2
2025 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2026 IP(src=server1.ip4, dst=server2_nat_ip) /
2027 TCP(sport=server_tcp_port, dport=self.tcp_port_out))
2029 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2030 IP(src=server1.ip4, dst=server2_nat_ip) /
2031 UDP(sport=server_udp_port, dport=self.udp_port_out))
2033 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2034 IP(src=server1.ip4, dst=server2_nat_ip) /
2035 ICMP(id=self.icmp_id_out, type='echo-reply'))
2037 self.pg0.add_stream(pkts)
2038 self.pg_enable_capture(self.pg_interfaces)
2040 capture = self.pg0.get_capture(len(pkts))
2041 for packet in capture:
2043 self.assertEqual(packet[IP].src, server1_nat_ip)
2044 self.assertEqual(packet[IP].dst, server2.ip4)
2045 if packet.haslayer(TCP):
2046 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
2047 self.assertEqual(packet[TCP].sport, server_tcp_port)
2048 self.assert_packet_checksums_valid(packet)
2049 elif packet.haslayer(UDP):
2050 self.assertEqual(packet[UDP].dport, self.udp_port_in)
2051 self.assertEqual(packet[UDP].sport, server_udp_port)
2053 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
2055 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2058 def test_hairpinning_avoid_inf_loop(self):
2059 """ NAT44EI hairpinning - 1:1 NAPT avoid infinite loop """
2061 host = self.pg0.remote_hosts[0]
2062 server = self.pg0.remote_hosts[1]
2065 server_in_port = 5678
2066 server_out_port = 8765
2068 self.nat44_add_address(self.nat_addr)
2069 flags = self.config_flags.NAT44_EI_IF_INSIDE
2070 self.vapi.nat44_ei_interface_add_del_feature(
2071 sw_if_index=self.pg0.sw_if_index,
2072 flags=flags, is_add=1)
2073 self.vapi.nat44_ei_interface_add_del_feature(
2074 sw_if_index=self.pg1.sw_if_index,
2077 # add static mapping for server
2078 self.nat44_add_static_mapping(server.ip4, self.nat_addr,
2079 server_in_port, server_out_port,
2080 proto=IP_PROTOS.tcp)
2082 # add another static mapping that maps pg0.local_ip4 address to itself
2083 self.nat44_add_static_mapping(self.pg0.local_ip4, self.pg0.local_ip4)
2085 # send packet from host to VPP (the packet should get dropped)
2086 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2087 IP(src=host.ip4, dst=self.pg0.local_ip4) /
2088 TCP(sport=host_in_port, dport=server_out_port))
2089 self.pg0.add_stream(p)
2090 self.pg_enable_capture(self.pg_interfaces)
2092 # Here VPP used to crash due to an infinite loop
2094 cnt = self.statistics['/nat44-ei/hairpinning']
2095 # send packet from host to server
2096 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2097 IP(src=host.ip4, dst=self.nat_addr) /
2098 TCP(sport=host_in_port, dport=server_out_port))
2099 self.pg0.add_stream(p)
2100 self.pg_enable_capture(self.pg_interfaces)
2102 capture = self.pg0.get_capture(1)
2107 self.assertEqual(ip.src, self.nat_addr)
2108 self.assertEqual(ip.dst, server.ip4)
2109 self.assertNotEqual(tcp.sport, host_in_port)
2110 self.assertEqual(tcp.dport, server_in_port)
2111 self.assert_packet_checksums_valid(p)
2112 host_out_port = tcp.sport
2114 self.logger.error(ppp("Unexpected or invalid packet:", p))
2117 after = self.statistics['/nat44-ei/hairpinning']
2118 if_idx = self.pg0.sw_if_index
2119 self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(), 1)
2121 # send reply from server to host
2122 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
2123 IP(src=server.ip4, dst=self.nat_addr) /
2124 TCP(sport=server_in_port, dport=host_out_port))
2125 self.pg0.add_stream(p)
2126 self.pg_enable_capture(self.pg_interfaces)
2128 capture = self.pg0.get_capture(1)
2133 self.assertEqual(ip.src, self.nat_addr)
2134 self.assertEqual(ip.dst, host.ip4)
2135 self.assertEqual(tcp.sport, server_out_port)
2136 self.assertEqual(tcp.dport, host_in_port)
2137 self.assert_packet_checksums_valid(p)
2139 self.logger.error(ppp("Unexpected or invalid packet:", p))
2142 after = self.statistics['/nat44-ei/hairpinning']
2143 if_idx = self.pg0.sw_if_index
2144 self.assertEqual(after[:, if_idx].sum() - cnt[:, if_idx].sum(),
2145 2+(1 if self.vpp_worker_count > 0 else 0))
2147 def test_interface_addr(self):
2148 """ NAT44EI acquire addresses from interface """
2149 self.vapi.nat44_ei_add_del_interface_addr(
2151 sw_if_index=self.pg7.sw_if_index)
2153 # no address in NAT pool
2154 addresses = self.vapi.nat44_ei_address_dump()
2155 self.assertEqual(0, len(addresses))
2157 # configure interface address and check NAT address pool
2158 self.pg7.config_ip4()
2159 addresses = self.vapi.nat44_ei_address_dump()
2160 self.assertEqual(1, len(addresses))
2161 self.assertEqual(str(addresses[0].ip_address), self.pg7.local_ip4)
2163 # remove interface address and check NAT address pool
2164 self.pg7.unconfig_ip4()
2165 addresses = self.vapi.nat44_ei_address_dump()
2166 self.assertEqual(0, len(addresses))
2168 def test_interface_addr_static_mapping(self):
2169 """ NAT44EI Static mapping with addresses from interface """
2172 self.vapi.nat44_ei_add_del_interface_addr(
2174 sw_if_index=self.pg7.sw_if_index)
2175 self.nat44_add_static_mapping(
2177 external_sw_if_index=self.pg7.sw_if_index,
2180 # static mappings with external interface
2181 static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2182 self.assertEqual(1, len(static_mappings))
2183 self.assertEqual(self.pg7.sw_if_index,
2184 static_mappings[0].external_sw_if_index)
2185 self.assertEqual(static_mappings[0].tag, tag)
2187 # configure interface address and check static mappings
2188 self.pg7.config_ip4()
2189 static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2190 self.assertEqual(2, len(static_mappings))
2192 for sm in static_mappings:
2193 if sm.external_sw_if_index == 0xFFFFFFFF:
2194 self.assertEqual(str(sm.external_ip_address),
2196 self.assertEqual(sm.tag, tag)
2198 self.assertTrue(resolved)
2200 # remove interface address and check static mappings
2201 self.pg7.unconfig_ip4()
2202 static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2203 self.assertEqual(1, len(static_mappings))
2204 self.assertEqual(self.pg7.sw_if_index,
2205 static_mappings[0].external_sw_if_index)
2206 self.assertEqual(static_mappings[0].tag, tag)
2208 # configure interface address again and check static mappings
2209 self.pg7.config_ip4()
2210 static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2211 self.assertEqual(2, len(static_mappings))
2213 for sm in static_mappings:
2214 if sm.external_sw_if_index == 0xFFFFFFFF:
2215 self.assertEqual(str(sm.external_ip_address),
2217 self.assertEqual(sm.tag, tag)
2219 self.assertTrue(resolved)
2221 # remove static mapping
2222 self.nat44_add_static_mapping(
2224 external_sw_if_index=self.pg7.sw_if_index,
2227 static_mappings = self.vapi.nat44_ei_static_mapping_dump()
2228 self.assertEqual(0, len(static_mappings))
2230 def test_interface_addr_identity_nat(self):
2231 """ NAT44EI Identity NAT with addresses from interface """
2234 self.vapi.nat44_ei_add_del_interface_addr(
2236 sw_if_index=self.pg7.sw_if_index)
2237 self.vapi.nat44_ei_add_del_identity_mapping(
2239 sw_if_index=self.pg7.sw_if_index,
2241 protocol=IP_PROTOS.tcp,
2244 # identity mappings with external interface
2245 identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2246 self.assertEqual(1, len(identity_mappings))
2247 self.assertEqual(self.pg7.sw_if_index,
2248 identity_mappings[0].sw_if_index)
2250 # configure interface address and check identity mappings
2251 self.pg7.config_ip4()
2252 identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2254 self.assertEqual(2, len(identity_mappings))
2255 for sm in identity_mappings:
2256 if sm.sw_if_index == 0xFFFFFFFF:
2257 self.assertEqual(str(identity_mappings[0].ip_address),
2259 self.assertEqual(port, identity_mappings[0].port)
2260 self.assertEqual(IP_PROTOS.tcp, identity_mappings[0].protocol)
2262 self.assertTrue(resolved)
2264 # remove interface address and check identity mappings
2265 self.pg7.unconfig_ip4()
2266 identity_mappings = self.vapi.nat44_ei_identity_mapping_dump()
2267 self.assertEqual(1, len(identity_mappings))
2268 self.assertEqual(self.pg7.sw_if_index,
2269 identity_mappings[0].sw_if_index)
2271 def test_ipfix_nat44_sess(self):
2272 """ NAT44EI IPFIX logging NAT44EI session created/deleted """
2273 self.ipfix_domain_id = 10
2274 self.ipfix_src_port = 20202
2275 collector_port = 30303
2276 bind_layers(UDP, IPFIX, dport=30303)
2277 self.nat44_add_address(self.nat_addr)
2278 flags = self.config_flags.NAT44_EI_IF_INSIDE
2279 self.vapi.nat44_ei_interface_add_del_feature(
2280 sw_if_index=self.pg0.sw_if_index,
2281 flags=flags, is_add=1)
2282 self.vapi.nat44_ei_interface_add_del_feature(
2283 sw_if_index=self.pg1.sw_if_index,
2285 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2286 src_address=self.pg3.local_ip4,
2288 template_interval=10,
2289 collector_port=collector_port)
2290 self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2291 src_port=self.ipfix_src_port,
2294 pkts = self.create_stream_in(self.pg0, self.pg1)
2295 self.pg0.add_stream(pkts)
2296 self.pg_enable_capture(self.pg_interfaces)
2298 capture = self.pg1.get_capture(len(pkts))
2299 self.verify_capture_out(capture)
2300 self.nat44_add_address(self.nat_addr, is_add=0)
2301 self.vapi.ipfix_flush()
2302 capture = self.pg3.get_capture(7)
2303 ipfix = IPFIXDecoder()
2304 # first load template
2306 self.assertTrue(p.haslayer(IPFIX))
2307 self.assertEqual(p[IP].src, self.pg3.local_ip4)
2308 self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2309 self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2310 self.assertEqual(p[UDP].dport, collector_port)
2311 self.assertEqual(p[IPFIX].observationDomainID,
2312 self.ipfix_domain_id)
2313 if p.haslayer(Template):
2314 ipfix.add_template(p.getlayer(Template))
2315 # verify events in data set
2317 if p.haslayer(Data):
2318 data = ipfix.decode_data_set(p.getlayer(Set))
2319 self.verify_ipfix_nat44_ses(data)
2321 def test_ipfix_addr_exhausted(self):
2322 """ NAT44EI IPFIX logging NAT addresses exhausted """
2323 flags = self.config_flags.NAT44_EI_IF_INSIDE
2324 self.vapi.nat44_ei_interface_add_del_feature(
2325 sw_if_index=self.pg0.sw_if_index,
2326 flags=flags, is_add=1)
2327 self.vapi.nat44_ei_interface_add_del_feature(
2328 sw_if_index=self.pg1.sw_if_index,
2330 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2331 src_address=self.pg3.local_ip4,
2333 template_interval=10)
2334 self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2335 src_port=self.ipfix_src_port,
2338 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
2339 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2341 self.pg0.add_stream(p)
2342 self.pg_enable_capture(self.pg_interfaces)
2344 self.pg1.assert_nothing_captured()
2346 self.vapi.ipfix_flush()
2347 capture = self.pg3.get_capture(7)
2348 ipfix = IPFIXDecoder()
2349 # first load template
2351 self.assertTrue(p.haslayer(IPFIX))
2352 self.assertEqual(p[IP].src, self.pg3.local_ip4)
2353 self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2354 self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2355 self.assertEqual(p[UDP].dport, 4739)
2356 self.assertEqual(p[IPFIX].observationDomainID,
2357 self.ipfix_domain_id)
2358 if p.haslayer(Template):
2359 ipfix.add_template(p.getlayer(Template))
2360 # verify events in data set
2362 if p.haslayer(Data):
2363 data = ipfix.decode_data_set(p.getlayer(Set))
2364 self.verify_ipfix_addr_exhausted(data)
2366 def test_ipfix_max_sessions(self):
2367 """ NAT44EI IPFIX logging maximum session entries exceeded """
2368 self.nat44_add_address(self.nat_addr)
2369 flags = self.config_flags.NAT44_EI_IF_INSIDE
2370 self.vapi.nat44_ei_interface_add_del_feature(
2371 sw_if_index=self.pg0.sw_if_index,
2372 flags=flags, is_add=1)
2373 self.vapi.nat44_ei_interface_add_del_feature(
2374 sw_if_index=self.pg1.sw_if_index,
2377 max_sessions_per_thread = self.max_translations
2378 max_sessions = max(1, self.vpp_worker_count) * max_sessions_per_thread
2381 for i in range(0, max_sessions):
2382 src = "10.10.%u.%u" % ((i & 0xFF00) >> 8, i & 0xFF)
2383 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2384 IP(src=src, dst=self.pg1.remote_ip4) /
2387 self.pg0.add_stream(pkts)
2388 self.pg_enable_capture(self.pg_interfaces)
2391 self.pg1.get_capture(max_sessions)
2392 self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4,
2393 src_address=self.pg3.local_ip4,
2395 template_interval=10)
2396 self.vapi.nat44_ei_ipfix_enable_disable(domain_id=self.ipfix_domain_id,
2397 src_port=self.ipfix_src_port,
2400 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2401 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2403 self.pg0.add_stream(p)
2404 self.pg_enable_capture(self.pg_interfaces)
2406 self.pg1.assert_nothing_captured()
2408 self.vapi.ipfix_flush()
2409 capture = self.pg3.get_capture(7)
2410 ipfix = IPFIXDecoder()
2411 # first load template
2413 self.assertTrue(p.haslayer(IPFIX))
2414 self.assertEqual(p[IP].src, self.pg3.local_ip4)
2415 self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
2416 self.assertEqual(p[UDP].sport, self.ipfix_src_port)
2417 self.assertEqual(p[UDP].dport, 4739)
2418 self.assertEqual(p[IPFIX].observationDomainID,
2419 self.ipfix_domain_id)
2420 if p.haslayer(Template):
2421 ipfix.add_template(p.getlayer(Template))
2422 # verify events in data set
2424 if p.haslayer(Data):
2425 data = ipfix.decode_data_set(p.getlayer(Set))
2426 self.verify_ipfix_max_sessions(data, max_sessions_per_thread)
2428 def test_syslog_apmap(self):
2429 """ NAT44EI syslog address and port mapping creation and deletion """
2430 self.vapi.syslog_set_filter(
2431 self.SYSLOG_SEVERITY.SYSLOG_API_SEVERITY_INFO)
2432 self.vapi.syslog_set_sender(self.pg3.local_ip4, self.pg3.remote_ip4)
2433 self.nat44_add_address(self.nat_addr)
2434 flags = self.config_flags.NAT44_EI_IF_INSIDE
2435 self.vapi.nat44_ei_interface_add_del_feature(
2436 sw_if_index=self.pg0.sw_if_index,
2437 flags=flags, is_add=1)
2438 self.vapi.nat44_ei_interface_add_del_feature(
2439 sw_if_index=self.pg1.sw_if_index,
2442 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2443 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2444 TCP(sport=self.tcp_port_in, dport=20))
2445 self.pg0.add_stream(p)
2446 self.pg_enable_capture(self.pg_interfaces)
2448 capture = self.pg1.get_capture(1)
2449 self.tcp_port_out = capture[0][TCP].sport
2450 capture = self.pg3.get_capture(1)
2451 self.verify_syslog_apmap(capture[0][Raw].load)
2453 self.pg_enable_capture(self.pg_interfaces)
2455 self.nat44_add_address(self.nat_addr, is_add=0)
2456 capture = self.pg3.get_capture(1)
2457 self.verify_syslog_apmap(capture[0][Raw].load, False)
2459 def test_pool_addr_fib(self):
2460 """ NAT44EI add pool addresses to FIB """
2461 static_addr = '10.0.0.10'
2462 self.nat44_add_address(self.nat_addr)
2463 flags = self.config_flags.NAT44_EI_IF_INSIDE
2464 self.vapi.nat44_ei_interface_add_del_feature(
2465 sw_if_index=self.pg0.sw_if_index,
2466 flags=flags, is_add=1)
2467 self.vapi.nat44_ei_interface_add_del_feature(
2468 sw_if_index=self.pg1.sw_if_index,
2470 self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr)
2473 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2474 ARP(op=ARP.who_has, pdst=self.nat_addr,
2475 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2476 self.pg1.add_stream(p)
2477 self.pg_enable_capture(self.pg_interfaces)
2479 capture = self.pg1.get_capture(1)
2480 self.assertTrue(capture[0].haslayer(ARP))
2481 self.assertTrue(capture[0][ARP].op, ARP.is_at)
2484 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2485 ARP(op=ARP.who_has, pdst=static_addr,
2486 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2487 self.pg1.add_stream(p)
2488 self.pg_enable_capture(self.pg_interfaces)
2490 capture = self.pg1.get_capture(1)
2491 self.assertTrue(capture[0].haslayer(ARP))
2492 self.assertTrue(capture[0][ARP].op, ARP.is_at)
2494 # send ARP to non-NAT44EI interface
2495 p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2496 ARP(op=ARP.who_has, pdst=self.nat_addr,
2497 psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
2498 self.pg2.add_stream(p)
2499 self.pg_enable_capture(self.pg_interfaces)
2501 self.pg1.assert_nothing_captured()
2503 # remove addresses and verify
2504 self.nat44_add_address(self.nat_addr, is_add=0)
2505 self.nat44_add_static_mapping(self.pg0.remote_ip4, static_addr,
2508 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2509 ARP(op=ARP.who_has, pdst=self.nat_addr,
2510 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2511 self.pg1.add_stream(p)
2512 self.pg_enable_capture(self.pg_interfaces)
2514 self.pg1.assert_nothing_captured()
2516 p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
2517 ARP(op=ARP.who_has, pdst=static_addr,
2518 psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
2519 self.pg1.add_stream(p)
2520 self.pg_enable_capture(self.pg_interfaces)
2522 self.pg1.assert_nothing_captured()
2524 def test_vrf_mode(self):
2525 """ NAT44EI tenant VRF aware address pool mode """
2529 nat_ip1 = "10.0.0.10"
2530 nat_ip2 = "10.0.0.11"
2532 self.pg0.unconfig_ip4()
2533 self.pg1.unconfig_ip4()
2534 self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
2535 self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
2536 self.pg0.set_table_ip4(vrf_id1)
2537 self.pg1.set_table_ip4(vrf_id2)
2538 self.pg0.config_ip4()
2539 self.pg1.config_ip4()
2540 self.pg0.resolve_arp()
2541 self.pg1.resolve_arp()
2543 self.nat44_add_address(nat_ip1, vrf_id=vrf_id1)
2544 self.nat44_add_address(nat_ip2, vrf_id=vrf_id2)
2545 flags = self.config_flags.NAT44_EI_IF_INSIDE
2546 self.vapi.nat44_ei_interface_add_del_feature(
2547 sw_if_index=self.pg0.sw_if_index,
2548 flags=flags, is_add=1)
2549 self.vapi.nat44_ei_interface_add_del_feature(
2550 sw_if_index=self.pg1.sw_if_index,
2551 flags=flags, is_add=1)
2552 self.vapi.nat44_ei_interface_add_del_feature(
2553 sw_if_index=self.pg2.sw_if_index,
2558 pkts = self.create_stream_in(self.pg0, self.pg2)
2559 self.pg0.add_stream(pkts)
2560 self.pg_enable_capture(self.pg_interfaces)
2562 capture = self.pg2.get_capture(len(pkts))
2563 self.verify_capture_out(capture, nat_ip1)
2566 pkts = self.create_stream_in(self.pg1, self.pg2)
2567 self.pg1.add_stream(pkts)
2568 self.pg_enable_capture(self.pg_interfaces)
2570 capture = self.pg2.get_capture(len(pkts))
2571 self.verify_capture_out(capture, nat_ip2)
2574 self.pg0.unconfig_ip4()
2575 self.pg1.unconfig_ip4()
2576 self.pg0.set_table_ip4(0)
2577 self.pg1.set_table_ip4(0)
2578 self.pg0.config_ip4()
2579 self.pg1.config_ip4()
2580 self.pg0.resolve_arp()
2581 self.pg1.resolve_arp()
2582 self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id1})
2583 self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id2})
2585 def test_vrf_feature_independent(self):
2586 """ NAT44EI tenant VRF independent address pool mode """
2588 nat_ip1 = "10.0.0.10"
2589 nat_ip2 = "10.0.0.11"
2591 self.nat44_add_address(nat_ip1)
2592 self.nat44_add_address(nat_ip2, vrf_id=99)
2593 flags = self.config_flags.NAT44_EI_IF_INSIDE
2594 self.vapi.nat44_ei_interface_add_del_feature(
2595 sw_if_index=self.pg0.sw_if_index,
2596 flags=flags, is_add=1)
2597 self.vapi.nat44_ei_interface_add_del_feature(
2598 sw_if_index=self.pg1.sw_if_index,
2599 flags=flags, is_add=1)
2600 self.vapi.nat44_ei_interface_add_del_feature(
2601 sw_if_index=self.pg2.sw_if_index,
2605 pkts = self.create_stream_in(self.pg0, self.pg2)
2606 self.pg0.add_stream(pkts)
2607 self.pg_enable_capture(self.pg_interfaces)
2609 capture = self.pg2.get_capture(len(pkts))
2610 self.verify_capture_out(capture, nat_ip1)
2613 pkts = self.create_stream_in(self.pg1, self.pg2)
2614 self.pg1.add_stream(pkts)
2615 self.pg_enable_capture(self.pg_interfaces)
2617 capture = self.pg2.get_capture(len(pkts))
2618 self.verify_capture_out(capture, nat_ip1)
2620 def test_dynamic_ipless_interfaces(self):
2621 """ NAT44EI interfaces without configured IP address """
2622 self.create_routes_and_neigbors()
2623 self.nat44_add_address(self.nat_addr)
2624 flags = self.config_flags.NAT44_EI_IF_INSIDE
2625 self.vapi.nat44_ei_interface_add_del_feature(
2626 sw_if_index=self.pg7.sw_if_index,
2627 flags=flags, is_add=1)
2628 self.vapi.nat44_ei_interface_add_del_feature(
2629 sw_if_index=self.pg8.sw_if_index,
2633 pkts = self.create_stream_in(self.pg7, self.pg8)
2634 self.pg7.add_stream(pkts)
2635 self.pg_enable_capture(self.pg_interfaces)
2637 capture = self.pg8.get_capture(len(pkts))
2638 self.verify_capture_out(capture)
2641 pkts = self.create_stream_out(self.pg8, self.nat_addr)
2642 self.pg8.add_stream(pkts)
2643 self.pg_enable_capture(self.pg_interfaces)
2645 capture = self.pg7.get_capture(len(pkts))
2646 self.verify_capture_in(capture, self.pg7)
2648 def test_static_ipless_interfaces(self):
2649 """ NAT44EI interfaces without configured IP address - 1:1 NAT """
2651 self.create_routes_and_neigbors()
2652 self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr)
2653 flags = self.config_flags.NAT44_EI_IF_INSIDE
2654 self.vapi.nat44_ei_interface_add_del_feature(
2655 sw_if_index=self.pg7.sw_if_index,
2656 flags=flags, is_add=1)
2657 self.vapi.nat44_ei_interface_add_del_feature(
2658 sw_if_index=self.pg8.sw_if_index,
2662 pkts = self.create_stream_out(self.pg8)
2663 self.pg8.add_stream(pkts)
2664 self.pg_enable_capture(self.pg_interfaces)
2666 capture = self.pg7.get_capture(len(pkts))
2667 self.verify_capture_in(capture, self.pg7)
2670 pkts = self.create_stream_in(self.pg7, self.pg8)
2671 self.pg7.add_stream(pkts)
2672 self.pg_enable_capture(self.pg_interfaces)
2674 capture = self.pg8.get_capture(len(pkts))
2675 self.verify_capture_out(capture, self.nat_addr, True)
2677 def test_static_with_port_ipless_interfaces(self):
2678 """ NAT44EI interfaces without configured IP address - 1:1 NAPT """
2680 self.tcp_port_out = 30606
2681 self.udp_port_out = 30607
2682 self.icmp_id_out = 30608
2684 self.create_routes_and_neigbors()
2685 self.nat44_add_address(self.nat_addr)
2686 self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2687 self.tcp_port_in, self.tcp_port_out,
2688 proto=IP_PROTOS.tcp)
2689 self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2690 self.udp_port_in, self.udp_port_out,
2691 proto=IP_PROTOS.udp)
2692 self.nat44_add_static_mapping(self.pg7.remote_ip4, self.nat_addr,
2693 self.icmp_id_in, self.icmp_id_out,
2694 proto=IP_PROTOS.icmp)
2695 flags = self.config_flags.NAT44_EI_IF_INSIDE
2696 self.vapi.nat44_ei_interface_add_del_feature(
2697 sw_if_index=self.pg7.sw_if_index,
2698 flags=flags, is_add=1)
2699 self.vapi.nat44_ei_interface_add_del_feature(
2700 sw_if_index=self.pg8.sw_if_index,
2704 pkts = self.create_stream_out(self.pg8)
2705 self.pg8.add_stream(pkts)
2706 self.pg_enable_capture(self.pg_interfaces)
2708 capture = self.pg7.get_capture(len(pkts))
2709 self.verify_capture_in(capture, self.pg7)
2712 pkts = self.create_stream_in(self.pg7, self.pg8)
2713 self.pg7.add_stream(pkts)
2714 self.pg_enable_capture(self.pg_interfaces)
2716 capture = self.pg8.get_capture(len(pkts))
2717 self.verify_capture_out(capture)
2719 def test_static_unknown_proto(self):
2720 """ NAT44EI 1:1 translate packet with unknown protocol """
2721 nat_ip = "10.0.0.10"
2722 self.nat44_add_static_mapping(self.pg0.remote_ip4, nat_ip)
2723 flags = self.config_flags.NAT44_EI_IF_INSIDE
2724 self.vapi.nat44_ei_interface_add_del_feature(
2725 sw_if_index=self.pg0.sw_if_index,
2726 flags=flags, is_add=1)
2727 self.vapi.nat44_ei_interface_add_del_feature(
2728 sw_if_index=self.pg1.sw_if_index,
2732 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
2733 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
2735 IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2736 TCP(sport=1234, dport=1234))
2737 self.pg0.add_stream(p)
2738 self.pg_enable_capture(self.pg_interfaces)
2740 p = self.pg1.get_capture(1)
2743 self.assertEqual(packet[IP].src, nat_ip)
2744 self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
2745 self.assertEqual(packet.haslayer(GRE), 1)
2746 self.assert_packet_checksums_valid(packet)
2748 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2752 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
2753 IP(src=self.pg1.remote_ip4, dst=nat_ip) /
2755 IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2756 TCP(sport=1234, dport=1234))
2757 self.pg1.add_stream(p)
2758 self.pg_enable_capture(self.pg_interfaces)
2760 p = self.pg0.get_capture(1)
2763 self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
2764 self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
2765 self.assertEqual(packet.haslayer(GRE), 1)
2766 self.assert_packet_checksums_valid(packet)
2768 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2771 def test_hairpinning_static_unknown_proto(self):
2772 """ NAT44EI 1:1 translate packet with unknown protocol - hairpinning
2775 host = self.pg0.remote_hosts[0]
2776 server = self.pg0.remote_hosts[1]
2778 host_nat_ip = "10.0.0.10"
2779 server_nat_ip = "10.0.0.11"
2781 self.nat44_add_static_mapping(host.ip4, host_nat_ip)
2782 self.nat44_add_static_mapping(server.ip4, server_nat_ip)
2783 flags = self.config_flags.NAT44_EI_IF_INSIDE
2784 self.vapi.nat44_ei_interface_add_del_feature(
2785 sw_if_index=self.pg0.sw_if_index,
2786 flags=flags, is_add=1)
2787 self.vapi.nat44_ei_interface_add_del_feature(
2788 sw_if_index=self.pg1.sw_if_index,
2792 p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
2793 IP(src=host.ip4, dst=server_nat_ip) /
2795 IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
2796 TCP(sport=1234, dport=1234))
2797 self.pg0.add_stream(p)
2798 self.pg_enable_capture(self.pg_interfaces)
2800 p = self.pg0.get_capture(1)
2803 self.assertEqual(packet[IP].src, host_nat_ip)
2804 self.assertEqual(packet[IP].dst, server.ip4)
2805 self.assertEqual(packet.haslayer(GRE), 1)
2806 self.assert_packet_checksums_valid(packet)
2808 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2812 p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
2813 IP(src=server.ip4, dst=host_nat_ip) /
2815 IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
2816 TCP(sport=1234, dport=1234))
2817 self.pg0.add_stream(p)
2818 self.pg_enable_capture(self.pg_interfaces)
2820 p = self.pg0.get_capture(1)
2823 self.assertEqual(packet[IP].src, server_nat_ip)
2824 self.assertEqual(packet[IP].dst, host.ip4)
2825 self.assertEqual(packet.haslayer(GRE), 1)
2826 self.assert_packet_checksums_valid(packet)
2828 self.logger.error(ppp("Unexpected or invalid packet:", packet))
2831 def test_output_feature(self):
2832 """ NAT44EI output feature (in2out postrouting) """
2833 self.nat44_add_address(self.nat_addr)
2834 flags = self.config_flags.NAT44_EI_IF_INSIDE
2835 self.vapi.nat44_ei_interface_add_del_output_feature(
2836 is_add=1, flags=flags,
2837 sw_if_index=self.pg0.sw_if_index)
2838 self.vapi.nat44_ei_interface_add_del_output_feature(
2839 is_add=1, flags=flags,
2840 sw_if_index=self.pg1.sw_if_index)
2841 self.vapi.nat44_ei_interface_add_del_output_feature(
2843 sw_if_index=self.pg3.sw_if_index)
2846 pkts = self.create_stream_in(self.pg0, self.pg3)
2847 self.pg0.add_stream(pkts)
2848 self.pg_enable_capture(self.pg_interfaces)
2850 capture = self.pg3.get_capture(len(pkts))
2851 self.verify_capture_out(capture)
2854 pkts = self.create_stream_out(self.pg3)
2855 self.pg3.add_stream(pkts)
2856 self.pg_enable_capture(self.pg_interfaces)
2858 capture = self.pg0.get_capture(len(pkts))
2859 self.verify_capture_in(capture, self.pg0)
2861 # from non-NAT interface to NAT inside interface
2862 pkts = self.create_stream_in(self.pg2, self.pg0)
2863 self.pg2.add_stream(pkts)
2864 self.pg_enable_capture(self.pg_interfaces)
2866 capture = self.pg0.get_capture(len(pkts))
2867 self.verify_capture_no_translation(capture, self.pg2, self.pg0)
2869 def test_output_feature_vrf_aware(self):
2870 """ NAT44EI output feature VRF aware (in2out postrouting) """
2871 nat_ip_vrf10 = "10.0.0.10"
2872 nat_ip_vrf20 = "10.0.0.20"
2874 r1 = VppIpRoute(self, self.pg3.remote_ip4, 32,
2875 [VppRoutePath(self.pg3.remote_ip4,
2876 self.pg3.sw_if_index)],
2878 r2 = VppIpRoute(self, self.pg3.remote_ip4, 32,
2879 [VppRoutePath(self.pg3.remote_ip4,
2880 self.pg3.sw_if_index)],
2885 self.nat44_add_address(nat_ip_vrf10, vrf_id=10)
2886 self.nat44_add_address(nat_ip_vrf20, vrf_id=20)
2887 flags = self.config_flags.NAT44_EI_IF_INSIDE
2888 self.vapi.nat44_ei_interface_add_del_output_feature(
2889 is_add=1, flags=flags,
2890 sw_if_index=self.pg4.sw_if_index)
2891 self.vapi.nat44_ei_interface_add_del_output_feature(
2892 is_add=1, flags=flags,
2893 sw_if_index=self.pg6.sw_if_index)
2894 self.vapi.nat44_ei_interface_add_del_output_feature(
2896 sw_if_index=self.pg3.sw_if_index)
2899 pkts = self.create_stream_in(self.pg4, self.pg3)
2900 self.pg4.add_stream(pkts)
2901 self.pg_enable_capture(self.pg_interfaces)
2903 capture = self.pg3.get_capture(len(pkts))
2904 self.verify_capture_out(capture, nat_ip=nat_ip_vrf10)
2907 pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10)
2908 self.pg3.add_stream(pkts)
2909 self.pg_enable_capture(self.pg_interfaces)
2911 capture = self.pg4.get_capture(len(pkts))
2912 self.verify_capture_in(capture, self.pg4)
2915 pkts = self.create_stream_in(self.pg6, self.pg3)
2916 self.pg6.add_stream(pkts)
2917 self.pg_enable_capture(self.pg_interfaces)
2919 capture = self.pg3.get_capture(len(pkts))
2920 self.verify_capture_out(capture, nat_ip=nat_ip_vrf20)
2923 pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20)
2924 self.pg3.add_stream(pkts)
2925 self.pg_enable_capture(self.pg_interfaces)
2927 capture = self.pg6.get_capture(len(pkts))
2928 self.verify_capture_in(capture, self.pg6)
2930 def test_output_feature_hairpinning(self):
2931 """ NAT44EI output feature hairpinning (in2out postrouting) """
2932 host = self.pg0.remote_hosts[0]
2933 server = self.pg0.remote_hosts[1]
2936 server_in_port = 5678
2937 server_out_port = 8765
2939 self.nat44_add_address(self.nat_addr)
2940 flags = self.config_flags.NAT44_EI_IF_INSIDE
2941 self.vapi.nat44_ei_interface_add_del_output_feature(
2942 is_add=1, flags=flags,
2943 sw_if_index=self.pg0.sw_if_index)
2944 self.vapi.nat44_ei_interface_add_del_output_feature(
2946 sw_if_index=self.pg1.sw_if_index)
2948 # add static mapping for server
2949 self.nat44_add_static_mapping(server.ip4, self.nat_addr,
2950 server_in_port, server_out_port,
2951 proto=IP_PROTOS.tcp)
2953 # send packet from host to server
2954 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
2955 IP(src=host.ip4, dst=self.nat_addr) /
2956 TCP(sport=host_in_port, dport=server_out_port))
2957 self.pg0.add_stream(p)
2958 self.pg_enable_capture(self.pg_interfaces)
2960 capture = self.pg0.get_capture(1)
2965 self.assertEqual(ip.src, self.nat_addr)
2966 self.assertEqual(ip.dst, server.ip4)
2967 self.assertNotEqual(tcp.sport, host_in_port)
2968 self.assertEqual(tcp.dport, server_in_port)
2969 self.assert_packet_checksums_valid(p)
2970 host_out_port = tcp.sport
2972 self.logger.error(ppp("Unexpected or invalid packet:", p))
2975 # send reply from server to host
2976 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
2977 IP(src=server.ip4, dst=self.nat_addr) /
2978 TCP(sport=server_in_port, dport=host_out_port))
2979 self.pg0.add_stream(p)
2980 self.pg_enable_capture(self.pg_interfaces)
2982 capture = self.pg0.get_capture(1)
2987 self.assertEqual(ip.src, self.nat_addr)
2988 self.assertEqual(ip.dst, host.ip4)
2989 self.assertEqual(tcp.sport, server_out_port)
2990 self.assertEqual(tcp.dport, host_in_port)
2991 self.assert_packet_checksums_valid(p)
2993 self.logger.error(ppp("Unexpected or invalid packet:", p))
2996 def test_one_armed_nat44(self):
2997 """ NAT44EI One armed NAT """
2998 remote_host = self.pg9.remote_hosts[0]
2999 local_host = self.pg9.remote_hosts[1]
3002 self.nat44_add_address(self.nat_addr)
3003 flags = self.config_flags.NAT44_EI_IF_INSIDE
3004 self.vapi.nat44_ei_interface_add_del_feature(
3005 sw_if_index=self.pg9.sw_if_index,
3007 self.vapi.nat44_ei_interface_add_del_feature(
3008 sw_if_index=self.pg9.sw_if_index,
3009 flags=flags, is_add=1)
3012 p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3013 IP(src=local_host.ip4, dst=remote_host.ip4) /
3014 TCP(sport=12345, dport=80))
3015 self.pg9.add_stream(p)
3016 self.pg_enable_capture(self.pg_interfaces)
3018 capture = self.pg9.get_capture(1)
3023 self.assertEqual(ip.src, self.nat_addr)
3024 self.assertEqual(ip.dst, remote_host.ip4)
3025 self.assertNotEqual(tcp.sport, 12345)
3026 external_port = tcp.sport
3027 self.assertEqual(tcp.dport, 80)
3028 self.assert_packet_checksums_valid(p)
3030 self.logger.error(ppp("Unexpected or invalid packet:", p))
3034 p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
3035 IP(src=remote_host.ip4, dst=self.nat_addr) /
3036 TCP(sport=80, dport=external_port))
3037 self.pg9.add_stream(p)
3038 self.pg_enable_capture(self.pg_interfaces)
3040 capture = self.pg9.get_capture(1)
3045 self.assertEqual(ip.src, remote_host.ip4)
3046 self.assertEqual(ip.dst, local_host.ip4)
3047 self.assertEqual(tcp.sport, 80)
3048 self.assertEqual(tcp.dport, 12345)
3049 self.assert_packet_checksums_valid(p)
3051 self.logger.error(ppp("Unexpected or invalid packet:", p))
3054 if self.vpp_worker_count > 1:
3055 node = "nat44-ei-handoff-classify"
3057 node = "nat44-ei-classify"
3059 err = self.statistics.get_err_counter('/err/%s/next in2out' % node)
3060 self.assertEqual(err, 1)
3061 err = self.statistics.get_err_counter('/err/%s/next out2in' % node)
3062 self.assertEqual(err, 1)
3064 def test_del_session(self):
3065 """ NAT44EI delete session """
3066 self.nat44_add_address(self.nat_addr)
3067 flags = self.config_flags.NAT44_EI_IF_INSIDE
3068 self.vapi.nat44_ei_interface_add_del_feature(
3069 sw_if_index=self.pg0.sw_if_index,
3070 flags=flags, is_add=1)
3071 self.vapi.nat44_ei_interface_add_del_feature(
3072 sw_if_index=self.pg1.sw_if_index,
3075 pkts = self.create_stream_in(self.pg0, self.pg1)
3076 self.pg0.add_stream(pkts)
3077 self.pg_enable_capture(self.pg_interfaces)
3079 self.pg1.get_capture(len(pkts))
3081 sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
3082 nsessions = len(sessions)
3084 self.vapi.nat44_ei_del_session(
3085 address=sessions[0].inside_ip_address,
3086 port=sessions[0].inside_port,
3087 protocol=sessions[0].protocol,
3088 flags=self.config_flags.NAT44_EI_IF_INSIDE)
3090 self.vapi.nat44_ei_del_session(
3091 address=sessions[1].outside_ip_address,
3092 port=sessions[1].outside_port,
3093 protocol=sessions[1].protocol)
3095 sessions = self.vapi.nat44_ei_user_session_dump(self.pg0.remote_ip4, 0)
3096 self.assertEqual(nsessions - len(sessions), 2)
3098 self.vapi.nat44_ei_del_session(
3099 address=sessions[0].inside_ip_address,
3100 port=sessions[0].inside_port,
3101 protocol=sessions[0].protocol,
3102 flags=self.config_flags.NAT44_EI_IF_INSIDE)
3104 self.verify_no_nat44_user()
3106 def test_frag_in_order(self):
3107 """ NAT44EI translate fragments arriving in order """
3109 self.nat44_add_address(self.nat_addr)
3110 flags = self.config_flags.NAT44_EI_IF_INSIDE
3111 self.vapi.nat44_ei_interface_add_del_feature(
3112 sw_if_index=self.pg0.sw_if_index,
3113 flags=flags, is_add=1)
3114 self.vapi.nat44_ei_interface_add_del_feature(
3115 sw_if_index=self.pg1.sw_if_index,
3118 self.frag_in_order(proto=IP_PROTOS.tcp)
3119 self.frag_in_order(proto=IP_PROTOS.udp)
3120 self.frag_in_order(proto=IP_PROTOS.icmp)
3122 def test_frag_forwarding(self):
3123 """ NAT44EI forwarding fragment test """
3124 self.vapi.nat44_ei_add_del_interface_addr(
3126 sw_if_index=self.pg1.sw_if_index)
3127 flags = self.config_flags.NAT44_EI_IF_INSIDE
3128 self.vapi.nat44_ei_interface_add_del_feature(
3129 sw_if_index=self.pg0.sw_if_index,
3130 flags=flags, is_add=1)
3131 self.vapi.nat44_ei_interface_add_del_feature(
3132 sw_if_index=self.pg1.sw_if_index,
3134 self.vapi.nat44_ei_forwarding_enable_disable(enable=1)
3136 data = b"A" * 16 + b"B" * 16 + b"C" * 3
3137 pkts = self.create_stream_frag(self.pg1,
3138 self.pg0.remote_ip4,
3142 proto=IP_PROTOS.udp)
3143 self.pg1.add_stream(pkts)
3144 self.pg_enable_capture(self.pg_interfaces)
3146 frags = self.pg0.get_capture(len(pkts))
3147 p = self.reass_frags_and_verify(frags,
3148 self.pg1.remote_ip4,
3149 self.pg0.remote_ip4)
3150 self.assertEqual(p[UDP].sport, 4789)
3151 self.assertEqual(p[UDP].dport, 4789)
3152 self.assertEqual(data, p[Raw].load)
3154 def test_reass_hairpinning(self):
3155 """ NAT44EI fragments hairpinning """
3157 server_addr = self.pg0.remote_hosts[1].ip4
3158 host_in_port = random.randint(1025, 65535)
3159 server_in_port = random.randint(1025, 65535)
3160 server_out_port = random.randint(1025, 65535)
3162 self.nat44_add_address(self.nat_addr)
3163 flags = self.config_flags.NAT44_EI_IF_INSIDE
3164 self.vapi.nat44_ei_interface_add_del_feature(
3165 sw_if_index=self.pg0.sw_if_index,
3166 flags=flags, is_add=1)
3167 self.vapi.nat44_ei_interface_add_del_feature(
3168 sw_if_index=self.pg1.sw_if_index,
3170 # add static mapping for server
3171 self.nat44_add_static_mapping(server_addr, self.nat_addr,
3174 proto=IP_PROTOS.tcp)
3175 self.nat44_add_static_mapping(server_addr, self.nat_addr,
3178 proto=IP_PROTOS.udp)
3179 self.nat44_add_static_mapping(server_addr, self.nat_addr)
3181 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3182 host_in_port, proto=IP_PROTOS.tcp)
3183 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3184 host_in_port, proto=IP_PROTOS.udp)
3185 self.reass_hairpinning(server_addr, server_in_port, server_out_port,
3186 host_in_port, proto=IP_PROTOS.icmp)
3188 def test_frag_out_of_order(self):
3189 """ NAT44EI translate fragments arriving out of order """
3191 self.nat44_add_address(self.nat_addr)
3192 flags = self.config_flags.NAT44_EI_IF_INSIDE
3193 self.vapi.nat44_ei_interface_add_del_feature(
3194 sw_if_index=self.pg0.sw_if_index,
3195 flags=flags, is_add=1)
3196 self.vapi.nat44_ei_interface_add_del_feature(
3197 sw_if_index=self.pg1.sw_if_index,
3200 self.frag_out_of_order(proto=IP_PROTOS.tcp)
3201 self.frag_out_of_order(proto=IP_PROTOS.udp)
3202 self.frag_out_of_order(proto=IP_PROTOS.icmp)
3204 def test_port_restricted(self):
3205 """ NAT44EI Port restricted NAT44EI (MAP-E CE) """
3206 self.nat44_add_address(self.nat_addr)
3207 flags = self.config_flags.NAT44_EI_IF_INSIDE
3208 self.vapi.nat44_ei_interface_add_del_feature(
3209 sw_if_index=self.pg0.sw_if_index,
3210 flags=flags, is_add=1)
3211 self.vapi.nat44_ei_interface_add_del_feature(
3212 sw_if_index=self.pg1.sw_if_index,
3214 self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=1,
3219 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3220 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3221 TCP(sport=4567, dport=22))
3222 self.pg0.add_stream(p)
3223 self.pg_enable_capture(self.pg_interfaces)
3225 capture = self.pg1.get_capture(1)
3230 self.assertEqual(ip.dst, self.pg1.remote_ip4)
3231 self.assertEqual(ip.src, self.nat_addr)
3232 self.assertEqual(tcp.dport, 22)
3233 self.assertNotEqual(tcp.sport, 4567)
3234 self.assertEqual((tcp.sport >> 6) & 63, 10)
3235 self.assert_packet_checksums_valid(p)
3237 self.logger.error(ppp("Unexpected or invalid packet:", p))
3240 def test_port_range(self):
3241 """ NAT44EI External address port range """
3242 self.nat44_add_address(self.nat_addr)
3243 flags = self.config_flags.NAT44_EI_IF_INSIDE
3244 self.vapi.nat44_ei_interface_add_del_feature(
3245 sw_if_index=self.pg0.sw_if_index,
3246 flags=flags, is_add=1)
3247 self.vapi.nat44_ei_interface_add_del_feature(
3248 sw_if_index=self.pg1.sw_if_index,
3250 self.vapi.nat44_ei_set_addr_and_port_alloc_alg(alg=2,
3255 for port in range(0, 5):
3256 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3257 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3258 TCP(sport=1125 + port))
3260 self.pg0.add_stream(pkts)
3261 self.pg_enable_capture(self.pg_interfaces)
3263 capture = self.pg1.get_capture(3)
3266 self.assertGreaterEqual(tcp.sport, 1025)
3267 self.assertLessEqual(tcp.sport, 1027)
3269 def test_multiple_outside_vrf(self):
3270 """ NAT44EI Multiple outside VRF """
3274 self.pg1.unconfig_ip4()
3275 self.pg2.unconfig_ip4()
3276 self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
3277 self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
3278 self.pg1.set_table_ip4(vrf_id1)
3279 self.pg2.set_table_ip4(vrf_id2)
3280 self.pg1.config_ip4()
3281 self.pg2.config_ip4()
3282 self.pg1.resolve_arp()
3283 self.pg2.resolve_arp()
3285 self.nat44_add_address(self.nat_addr)
3286 flags = self.config_flags.NAT44_EI_IF_INSIDE
3287 self.vapi.nat44_ei_interface_add_del_feature(
3288 sw_if_index=self.pg0.sw_if_index,
3289 flags=flags, is_add=1)
3290 self.vapi.nat44_ei_interface_add_del_feature(
3291 sw_if_index=self.pg1.sw_if_index,
3293 self.vapi.nat44_ei_interface_add_del_feature(
3294 sw_if_index=self.pg2.sw_if_index,
3299 pkts = self.create_stream_in(self.pg0, self.pg1)
3300 self.pg0.add_stream(pkts)
3301 self.pg_enable_capture(self.pg_interfaces)
3303 capture = self.pg1.get_capture(len(pkts))
3304 self.verify_capture_out(capture, self.nat_addr)
3306 pkts = self.create_stream_out(self.pg1, self.nat_addr)
3307 self.pg1.add_stream(pkts)
3308 self.pg_enable_capture(self.pg_interfaces)
3310 capture = self.pg0.get_capture(len(pkts))
3311 self.verify_capture_in(capture, self.pg0)
3313 self.tcp_port_in = 60303
3314 self.udp_port_in = 60304
3315 self.icmp_id_in = 60305
3318 pkts = self.create_stream_in(self.pg0, self.pg2)
3319 self.pg0.add_stream(pkts)
3320 self.pg_enable_capture(self.pg_interfaces)
3322 capture = self.pg2.get_capture(len(pkts))
3323 self.verify_capture_out(capture, self.nat_addr)
3325 pkts = self.create_stream_out(self.pg2, self.nat_addr)
3326 self.pg2.add_stream(pkts)
3327 self.pg_enable_capture(self.pg_interfaces)
3329 capture = self.pg0.get_capture(len(pkts))
3330 self.verify_capture_in(capture, self.pg0)
3333 self.nat44_add_address(self.nat_addr, is_add=0)
3334 self.pg1.unconfig_ip4()
3335 self.pg2.unconfig_ip4()
3336 self.pg1.set_table_ip4(0)
3337 self.pg2.set_table_ip4(0)
3338 self.pg1.config_ip4()
3339 self.pg2.config_ip4()
3340 self.pg1.resolve_arp()
3341 self.pg2.resolve_arp()
3343 def test_mss_clamping(self):
3344 """ NAT44EI TCP MSS clamping """
3345 self.nat44_add_address(self.nat_addr)
3346 flags = self.config_flags.NAT44_EI_IF_INSIDE
3347 self.vapi.nat44_ei_interface_add_del_feature(
3348 sw_if_index=self.pg0.sw_if_index,
3349 flags=flags, is_add=1)
3350 self.vapi.nat44_ei_interface_add_del_feature(
3351 sw_if_index=self.pg1.sw_if_index,
3354 p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
3355 IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
3356 TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
3357 flags="S", options=[('MSS', 1400)]))
3359 self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1000)
3360 self.pg0.add_stream(p)
3361 self.pg_enable_capture(self.pg_interfaces)
3363 capture = self.pg1.get_capture(1)
3364 # Negotiated MSS value greater than configured - changed
3365 self.verify_mss_value(capture[0], 1000)
3367 self.vapi.nat44_ei_set_mss_clamping(enable=0, mss_value=1500)
3368 self.pg0.add_stream(p)
3369 self.pg_enable_capture(self.pg_interfaces)
3371 capture = self.pg1.get_capture(1)
3372 # MSS clamping disabled - negotiated MSS unchanged
3373 self.verify_mss_value(capture[0], 1400)
3375 self.vapi.nat44_ei_set_mss_clamping(enable=1, mss_value=1500)
3376 self.pg0.add_stream(p)
3377 self.pg_enable_capture(self.pg_interfaces)
3379 capture = self.pg1.get_capture(1)
3380 # Negotiated MSS value smaller than configured - unchanged
3381 self.verify_mss_value(capture[0], 1400)
3383 def test_ha_send(self):
3384 """ NAT44EI Send HA session synchronization events (active) """
3385 flags = self.config_flags.NAT44_EI_IF_INSIDE
3386 self.vapi.nat44_ei_interface_add_del_feature(
3387 sw_if_index=self.pg0.sw_if_index,
3388 flags=flags, is_add=1)
3389 self.vapi.nat44_ei_interface_add_del_feature(
3390 sw_if_index=self.pg1.sw_if_index,
3392 self.nat44_add_address(self.nat_addr)
3394 self.vapi.nat44_ei_ha_set_listener(
3395 ip_address=self.pg3.local_ip4, port=12345, path_mtu=512)
3396 self.vapi.nat44_ei_ha_set_failover(
3397 ip_address=self.pg3.remote_ip4, port=12346,
3398 session_refresh_interval=10)
3399 bind_layers(UDP, HANATStateSync, sport=12345)
3402 pkts = self.create_stream_in(self.pg0, self.pg1)
3403 self.pg0.add_stream(pkts)
3404 self.pg_enable_capture(self.pg_interfaces)
3406 capture = self.pg1.get_capture(len(pkts))
3407 self.verify_capture_out(capture)
3408 # active send HA events
3409 self.vapi.nat44_ei_ha_flush()
3410 stats = self.statistics['/nat44-ei/ha/add-event-send']
3411 self.assertEqual(stats[:, 0].sum(), 3)
3412 capture = self.pg3.get_capture(1)
3414 self.assert_packet_checksums_valid(p)
3418 hanat = p[HANATStateSync]
3420 self.logger.error(ppp("Invalid packet:", p))
3423 self.assertEqual(ip.src, self.pg3.local_ip4)
3424 self.assertEqual(ip.dst, self.pg3.remote_ip4)
3425 self.assertEqual(udp.sport, 12345)
3426 self.assertEqual(udp.dport, 12346)
3427 self.assertEqual(hanat.version, 1)
3428 # self.assertEqual(hanat.thread_index, 0)
3429 self.assertEqual(hanat.count, 3)
3430 seq = hanat.sequence_number
3431 for event in hanat.events:
3432 self.assertEqual(event.event_type, 1)
3433 self.assertEqual(event.in_addr, self.pg0.remote_ip4)
3434 self.assertEqual(event.out_addr, self.nat_addr)
3435 self.assertEqual(event.fib_index, 0)
3437 # ACK received events
3438 ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3439 IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3440 UDP(sport=12346, dport=12345) /
3441 HANATStateSync(sequence_number=seq, flags='ACK',
3442 thread_index=hanat.thread_index))
3443 self.pg3.add_stream(ack)
3445 stats = self.statistics['/nat44-ei/ha/ack-recv']
3446 self.assertEqual(stats[:, 0].sum(), 1)
3448 # delete one session
3449 self.pg_enable_capture(self.pg_interfaces)
3450 self.vapi.nat44_ei_del_session(
3451 address=self.pg0.remote_ip4, port=self.tcp_port_in,
3452 protocol=IP_PROTOS.tcp, flags=self.config_flags.NAT44_EI_IF_INSIDE)
3453 self.vapi.nat44_ei_ha_flush()
3454 stats = self.statistics['/nat44-ei/ha/del-event-send']
3455 self.assertEqual(stats[:, 0].sum(), 1)
3456 capture = self.pg3.get_capture(1)
3459 hanat = p[HANATStateSync]
3461 self.logger.error(ppp("Invalid packet:", p))
3464 self.assertGreater(hanat.sequence_number, seq)
3466 # do not send ACK, active retry send HA event again
3467 self.pg_enable_capture(self.pg_interfaces)
3469 stats = self.statistics['/nat44-ei/ha/retry-count']
3470 self.assertEqual(stats[:, 0].sum(), 3)
3471 stats = self.statistics['/nat44-ei/ha/missed-count']
3472 self.assertEqual(stats[:, 0].sum(), 1)
3473 capture = self.pg3.get_capture(3)
3474 for packet in capture:
3475 self.assertEqual(packet, p)
3477 # session counters refresh
3478 pkts = self.create_stream_out(self.pg1)
3479 self.pg1.add_stream(pkts)
3480 self.pg_enable_capture(self.pg_interfaces)
3482 self.pg0.get_capture(2)
3483 self.vapi.nat44_ei_ha_flush()
3484 stats = self.statistics['/nat44-ei/ha/refresh-event-send']
3485 self.assertEqual(stats[:, 0].sum(), 2)
3486 capture = self.pg3.get_capture(1)
3488 self.assert_packet_checksums_valid(p)
3492 hanat = p[HANATStateSync]
3494 self.logger.error(ppp("Invalid packet:", p))
3497 self.assertEqual(ip.src, self.pg3.local_ip4)
3498 self.assertEqual(ip.dst, self.pg3.remote_ip4)
3499 self.assertEqual(udp.sport, 12345)
3500 self.assertEqual(udp.dport, 12346)
3501 self.assertEqual(hanat.version, 1)
3502 self.assertEqual(hanat.count, 2)
3503 seq = hanat.sequence_number
3504 for event in hanat.events:
3505 self.assertEqual(event.event_type, 3)
3506 self.assertEqual(event.out_addr, self.nat_addr)
3507 self.assertEqual(event.fib_index, 0)
3508 self.assertEqual(event.total_pkts, 2)
3509 self.assertGreater(event.total_bytes, 0)
3511 stats = self.statistics['/nat44-ei/ha/ack-recv']
3512 ack = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3513 IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3514 UDP(sport=12346, dport=12345) /
3515 HANATStateSync(sequence_number=seq, flags='ACK',
3516 thread_index=hanat.thread_index))
3517 self.pg3.add_stream(ack)
3519 stats = self.statistics['/nat44-ei/ha/ack-recv']
3520 self.assertEqual(stats[:, 0].sum(), 2)
3522 def test_ha_recv(self):
3523 """ NAT44EI Receive HA session synchronization events (passive) """
3524 self.nat44_add_address(self.nat_addr)
3525 flags = self.config_flags.NAT44_EI_IF_INSIDE
3526 self.vapi.nat44_ei_interface_add_del_feature(
3527 sw_if_index=self.pg0.sw_if_index,
3528 flags=flags, is_add=1)
3529 self.vapi.nat44_ei_interface_add_del_feature(
3530 sw_if_index=self.pg1.sw_if_index,
3532 self.vapi.nat44_ei_ha_set_listener(ip_address=self.pg3.local_ip4,
3533 port=12345, path_mtu=512)
3534 bind_layers(UDP, HANATStateSync, sport=12345)
3536 # this is a bit tricky - HA dictates thread index due to how it's
3537 # designed, but once we use HA to create a session, we also want
3538 # to pass a packet through said session. so the session must end
3539 # up on the correct thread from both directions - in2out (based on
3540 # IP address) and out2in (based on outside port)
3542 # first choose a thread index which is correct for IP
3543 thread_index = get_nat44_ei_in2out_worker_index(self.pg0.remote_ip4,
3544 self.vpp_worker_count)
3546 # now pick a port which is correct for given thread
3547 port_per_thread = int((0xffff-1024) / max(1, self.vpp_worker_count))
3548 self.tcp_port_out = 1024 + random.randint(1, port_per_thread)
3549 self.udp_port_out = 1024 + random.randint(1, port_per_thread)
3550 if self.vpp_worker_count > 0:
3551 self.tcp_port_out += port_per_thread * (thread_index - 1)
3552 self.udp_port_out += port_per_thread * (thread_index - 1)
3554 # send HA session add events to failover/passive
3555 p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3556 IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3557 UDP(sport=12346, dport=12345) /
3558 HANATStateSync(sequence_number=1, events=[
3559 Event(event_type='add', protocol='tcp',
3560 in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3561 in_port=self.tcp_port_in, out_port=self.tcp_port_out,
3562 eh_addr=self.pg1.remote_ip4,
3563 ehn_addr=self.pg1.remote_ip4,
3564 eh_port=self.tcp_external_port,
3565 ehn_port=self.tcp_external_port, fib_index=0),
3566 Event(event_type='add', protocol='udp',
3567 in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3568 in_port=self.udp_port_in, out_port=self.udp_port_out,
3569 eh_addr=self.pg1.remote_ip4,
3570 ehn_addr=self.pg1.remote_ip4,
3571 eh_port=self.udp_external_port,
3572 ehn_port=self.udp_external_port, fib_index=0)],
3573 thread_index=thread_index))
3575 self.pg3.add_stream(p)
3576 self.pg_enable_capture(self.pg_interfaces)
3579 capture = self.pg3.get_capture(1)
3582 hanat = p[HANATStateSync]
3584 self.logger.error(ppp("Invalid packet:", p))
3587 self.assertEqual(hanat.sequence_number, 1)
3588 self.assertEqual(hanat.flags, 'ACK')
3589 self.assertEqual(hanat.version, 1)
3590 self.assertEqual(hanat.thread_index, thread_index)
3591 stats = self.statistics['/nat44-ei/ha/ack-send']
3592 self.assertEqual(stats[:, 0].sum(), 1)
3593 stats = self.statistics['/nat44-ei/ha/add-event-recv']
3594 self.assertEqual(stats[:, 0].sum(), 2)
3595 users = self.statistics['/nat44-ei/total-users']
3596 self.assertEqual(users[:, 0].sum(), 1)
3597 sessions = self.statistics['/nat44-ei/total-sessions']
3598 self.assertEqual(sessions[:, 0].sum(), 2)
3599 users = self.vapi.nat44_ei_user_dump()
3600 self.assertEqual(len(users), 1)
3601 self.assertEqual(str(users[0].ip_address),
3602 self.pg0.remote_ip4)
3603 # there should be 2 sessions created by HA
3604 sessions = self.vapi.nat44_ei_user_session_dump(
3605 users[0].ip_address, users[0].vrf_id)
3606 self.assertEqual(len(sessions), 2)
3607 for session in sessions:
3608 self.assertEqual(str(session.inside_ip_address),
3609 self.pg0.remote_ip4)
3610 self.assertEqual(str(session.outside_ip_address),
3612 self.assertIn(session.inside_port,
3613 [self.tcp_port_in, self.udp_port_in])
3614 self.assertIn(session.outside_port,
3615 [self.tcp_port_out, self.udp_port_out])
3616 self.assertIn(session.protocol, [IP_PROTOS.tcp, IP_PROTOS.udp])
3618 # send HA session delete event to failover/passive
3619 p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3620 IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3621 UDP(sport=12346, dport=12345) /
3622 HANATStateSync(sequence_number=2, events=[
3623 Event(event_type='del', protocol='udp',
3624 in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3625 in_port=self.udp_port_in, out_port=self.udp_port_out,
3626 eh_addr=self.pg1.remote_ip4,
3627 ehn_addr=self.pg1.remote_ip4,
3628 eh_port=self.udp_external_port,
3629 ehn_port=self.udp_external_port, fib_index=0)],
3630 thread_index=thread_index))
3632 self.pg3.add_stream(p)
3633 self.pg_enable_capture(self.pg_interfaces)
3636 capture = self.pg3.get_capture(1)
3639 hanat = p[HANATStateSync]
3641 self.logger.error(ppp("Invalid packet:", p))
3644 self.assertEqual(hanat.sequence_number, 2)
3645 self.assertEqual(hanat.flags, 'ACK')
3646 self.assertEqual(hanat.version, 1)
3647 users = self.vapi.nat44_ei_user_dump()
3648 self.assertEqual(len(users), 1)
3649 self.assertEqual(str(users[0].ip_address),
3650 self.pg0.remote_ip4)
3651 # now we should have only 1 session, 1 deleted by HA
3652 sessions = self.vapi.nat44_ei_user_session_dump(users[0].ip_address,
3654 self.assertEqual(len(sessions), 1)
3655 stats = self.statistics['/nat44-ei/ha/del-event-recv']
3656 self.assertEqual(stats[:, 0].sum(), 1)
3658 stats = self.statistics.get_err_counter(
3659 '/err/nat44-ei-ha/pkts-processed')
3660 self.assertEqual(stats, 2)
3662 # send HA session refresh event to failover/passive
3663 p = (Ether(dst=self.pg3.local_mac, src=self.pg3.remote_mac) /
3664 IP(src=self.pg3.remote_ip4, dst=self.pg3.local_ip4) /
3665 UDP(sport=12346, dport=12345) /
3666 HANATStateSync(sequence_number=3, events=[
3667 Event(event_type='refresh', protocol='tcp',
3668 in_addr=self.pg0.remote_ip4, out_addr=self.nat_addr,
3669 in_port=self.tcp_port_in, out_port=self.tcp_port_out,
3670 eh_addr=self.pg1.remote_ip4,
3671 ehn_addr=self.pg1.remote_ip4,
3672 eh_port=self.tcp_external_port,
3673 ehn_port=self.tcp_external_port, fib_index=0,
3674 total_bytes=1024, total_pkts=2)],
3675 thread_index=thread_index))
3676 self.pg3.add_stream(p)
3677 self.pg_enable_capture(self.pg_interfaces)
3680 capture = self.pg3.get_capture(1)
3683 hanat = p[HANATStateSync]
3685 self.logger.error(ppp("Invalid packet:", p))
3688 self.assertEqual(hanat.sequence_number, 3)
3689 self.assertEqual(hanat.flags, 'ACK')
3690 self.assertEqual(hanat.version, 1)
3691 users = self.vapi.nat44_ei_user_dump()
3692 self.assertEqual(len(users), 1)
3693 self.assertEqual(str(users[0].ip_address),
3694 self.pg0.remote_ip4)
3695 sessions = self.vapi.nat44_ei_user_session_dump(
3696 users[0].ip_address, users[0].vrf_id)
3697 self.assertEqual(len(sessions), 1)
3698 session = sessions[0]
3699 self.assertEqual(session.total_bytes, 1024)
3700 self.assertEqual(session.total_pkts, 2)
3701 stats = self.statistics['/nat44-ei/ha/refresh-event-recv']
3702 self.assertEqual(stats[:, 0].sum(), 1)
3704 stats = self.statistics.get_err_counter(
3705 '/err/nat44-ei-ha/pkts-processed')
3706 self.assertEqual(stats, 3)
3708 # send packet to test session created by HA
3709 p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
3710 IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
3711 TCP(sport=self.tcp_external_port, dport=self.tcp_port_out))
3712 self.pg1.add_stream(p)
3713 self.pg_enable_capture(self.pg_interfaces)
3715 capture = self.pg0.get_capture(1)
3721 self.logger.error(ppp("Invalid packet:", p))
3724 self.assertEqual(ip.src, self.pg1.remote_ip4)
3725 self.assertEqual(ip.dst, self.pg0.remote_ip4)
3726 self.assertEqual(tcp.sport, self.tcp_external_port)
3727 self.assertEqual(tcp.dport, self.tcp_port_in)
3729 def reconfigure_frame_queue_nelts(self, frame_queue_nelts):
3730 self.vapi.nat44_ei_plugin_enable_disable(enable=0)
3731 self.vapi.nat44_ei_set_fq_options(frame_queue_nelts=frame_queue_nelts)
3732 # keep plugin configuration persistent
3733 self.plugin_enable()
3734 return self.vapi.nat44_ei_show_fq_options().frame_queue_nelts
3736 def test_set_frame_queue_nelts(self):
3737 """ NAT44 EI API test - worker handoff frame queue elements """
3738 self.assertEqual(self.reconfigure_frame_queue_nelts(512), 512)
3740 def show_commands_at_teardown(self):
3741 self.logger.info(self.vapi.cli("show nat44 ei timeouts"))
3742 self.logger.info(self.vapi.cli("show nat44 ei addresses"))
3743 self.logger.info(self.vapi.cli("show nat44 ei interfaces"))
3744 self.logger.info(self.vapi.cli("show nat44 ei static mappings"))
3745 self.logger.info(self.vapi.cli("show nat44 ei interface address"))
3746 self.logger.info(self.vapi.cli("show nat44 ei sessions detail"))
3747 self.logger.info(self.vapi.cli("show nat44 ei hash tables detail"))
3748 self.logger.info(self.vapi.cli("show nat44 ei ha"))
3750 self.vapi.cli("show nat44 ei addr-port-assignment-alg"))
3752 def test_outside_address_distribution(self):
3753 """ Outside address distribution based on source address """
3758 for i in range(1, x):
3760 nat_addresses.append(a)
3762 flags = self.config_flags.NAT44_EI_IF_INSIDE
3763 self.vapi.nat44_ei_interface_add_del_feature(
3764 sw_if_index=self.pg0.sw_if_index,
3765 flags=flags, is_add=1)
3766 self.vapi.nat44_ei_interface_add_del_feature(
3767 sw_if_index=self.pg1.sw_if_index,
3770 self.vapi.nat44_ei_add_del_address_range(
3771 first_ip_address=nat_addresses[0],
3772 last_ip_address=nat_addresses[-1],
3773 vrf_id=0xFFFFFFFF, is_add=1)
3775 self.pg0.generate_remote_hosts(x)
3779 info = self.create_packet_info(self.pg0, self.pg1)
3780 payload = self.info_to_payload(info)
3781 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
3782 IP(src=self.pg0.remote_hosts[i].ip4,
3783 dst=self.pg1.remote_ip4) /
3784 UDP(sport=7000+i, dport=8000+i) /
3789 self.pg0.add_stream(pkts)
3790 self.pg_enable_capture(self.pg_interfaces)
3792 recvd = self.pg1.get_capture(len(pkts))
3793 for p_recvd in recvd:
3794 payload_info = self.payload_to_info(p_recvd[Raw])
3795 packet_index = payload_info.index
3796 info = self._packet_infos[packet_index]
3797 self.assertTrue(info is not None)
3798 self.assertEqual(packet_index, info.index)
3800 packed = socket.inet_aton(p_sent[IP].src)
3801 numeric = struct.unpack("!L", packed)[0]
3802 numeric = socket.htonl(numeric)
3803 a = nat_addresses[(numeric-1) % len(nat_addresses)]
3806 "Invalid packet (src IP %s translated to %s, but expected %s)"
3807 % (p_sent[IP].src, p_recvd[IP].src, a))
3809 def test_default_user_sessions(self):
3810 """ NAT44EI default per-user session limit is used and reported """
3811 nat44_ei_config = self.vapi.nat44_ei_show_running_config()
3812 # a nonzero default should be reported for user_sessions
3813 self.assertNotEqual(nat44_ei_config.user_sessions, 0)
3816 class TestNAT44Out2InDPO(MethodHolder):
3817 """ NAT44EI Test Cases using out2in DPO """
3820 def setUpClass(cls):
3821 super(TestNAT44Out2InDPO, cls).setUpClass()
3822 cls.vapi.cli("set log class nat44-ei level debug")
3824 cls.tcp_port_in = 6303
3825 cls.tcp_port_out = 6303
3826 cls.udp_port_in = 6304
3827 cls.udp_port_out = 6304
3828 cls.icmp_id_in = 6305
3829 cls.icmp_id_out = 6305
3830 cls.nat_addr = '10.0.0.3'
3831 cls.dst_ip4 = '192.168.70.1'
3833 cls.create_pg_interfaces(range(2))
3836 cls.pg0.config_ip4()
3837 cls.pg0.resolve_arp()
3840 cls.pg1.config_ip6()
3841 cls.pg1.resolve_ndp()
3843 r1 = VppIpRoute(cls, "::", 0,
3844 [VppRoutePath(cls.pg1.remote_ip6,
3845 cls.pg1.sw_if_index)],
3850 super(TestNAT44Out2InDPO, self).setUp()
3851 flags = self.config_flags.NAT44_EI_OUT2IN_DPO
3852 self.vapi.nat44_ei_plugin_enable_disable(enable=1, flags=flags)
3855 super(TestNAT44Out2InDPO, self).tearDown()
3856 if not self.vpp_dead:
3857 self.vapi.nat44_ei_plugin_enable_disable(enable=0)
3858 self.vapi.cli("clear logging")
3860 def configure_xlat(self):
3861 self.dst_ip6_pfx = '1:2:3::'
3862 self.dst_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3864 self.dst_ip6_pfx_len = 96
3865 self.src_ip6_pfx = '4:5:6::'
3866 self.src_ip6_pfx_n = socket.inet_pton(socket.AF_INET6,
3868 self.src_ip6_pfx_len = 96
3869 self.vapi.map_add_domain(self.dst_ip6_pfx_n, self.dst_ip6_pfx_len,
3870 self.src_ip6_pfx_n, self.src_ip6_pfx_len,
3871 '\x00\x00\x00\x00', 0)
3873 @unittest.skip('Temporary disabled')
3874 def test_464xlat_ce(self):
3875 """ Test 464XLAT CE with NAT44EI """
3877 self.configure_xlat()
3879 flags = self.config_flags.NAT44_EI_IF_INSIDE
3880 self.vapi.nat44_ei_interface_add_del_feature(
3881 sw_if_index=self.pg0.sw_if_index,
3882 flags=flags, is_add=1)
3883 self.vapi.nat44_ei_add_del_address_range(
3884 first_ip_address=self.nat_addr_n,
3885 last_ip_address=self.nat_addr_n,
3886 vrf_id=0xFFFFFFFF, is_add=1)
3888 out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3889 self.dst_ip6_pfx_len)
3890 out_dst_ip6 = self.compose_ip6(self.nat_addr, self.src_ip6_pfx,
3891 self.src_ip6_pfx_len)
3894 pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3895 self.pg0.add_stream(pkts)
3896 self.pg_enable_capture(self.pg_interfaces)
3898 capture = self.pg1.get_capture(len(pkts))
3899 self.verify_capture_out_ip6(capture, nat_ip=out_dst_ip6,
3902 pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6,
3904 self.pg1.add_stream(pkts)
3905 self.pg_enable_capture(self.pg_interfaces)
3907 capture = self.pg0.get_capture(len(pkts))
3908 self.verify_capture_in(capture, self.pg0)
3910 self.vapi.nat44_ei_interface_add_del_feature(
3911 sw_if_index=self.pg0.sw_if_index,
3913 self.vapi.nat44_ei_add_del_address_range(
3914 first_ip_address=self.nat_addr_n,
3915 last_ip_address=self.nat_addr_n,
3918 @unittest.skip('Temporary disabled')
3919 def test_464xlat_ce_no_nat(self):
3920 """ Test 464XLAT CE without NAT44EI """
3922 self.configure_xlat()
3924 out_src_ip6 = self.compose_ip6(self.dst_ip4, self.dst_ip6_pfx,
3925 self.dst_ip6_pfx_len)
3926 out_dst_ip6 = self.compose_ip6(self.pg0.remote_ip4, self.src_ip6_pfx,
3927 self.src_ip6_pfx_len)
3929 pkts = self.create_stream_in(self.pg0, self.pg1, self.dst_ip4)
3930 self.pg0.add_stream(pkts)
3931 self.pg_enable_capture(self.pg_interfaces)
3933 capture = self.pg1.get_capture(len(pkts))
3934 self.verify_capture_out_ip6(capture, dst_ip=out_src_ip6,
3935 nat_ip=out_dst_ip6, same_port=True)
3937 pkts = self.create_stream_out_ip6(self.pg1, out_src_ip6, out_dst_ip6)
3938 self.pg1.add_stream(pkts)
3939 self.pg_enable_capture(self.pg_interfaces)
3941 capture = self.pg0.get_capture(len(pkts))
3942 self.verify_capture_in(capture, self.pg0)
3945 class TestNAT44EIMW(MethodHolder):
3946 """ NAT44EI Test Cases (multiple workers) """
3947 vpp_worker_count = 2
3948 max_translations = 10240
3952 def setUpClass(cls):
3953 super(TestNAT44EIMW, cls).setUpClass()
3954 cls.vapi.cli("set log class nat level debug")
3956 cls.tcp_port_in = 6303
3957 cls.tcp_port_out = 6303
3958 cls.udp_port_in = 6304
3959 cls.udp_port_out = 6304
3960 cls.icmp_id_in = 6305
3961 cls.icmp_id_out = 6305
3962 cls.nat_addr = '10.0.0.3'
3963 cls.ipfix_src_port = 4739
3964 cls.ipfix_domain_id = 1
3965 cls.tcp_external_port = 80
3966 cls.udp_external_port = 69
3968 cls.create_pg_interfaces(range(10))
3969 cls.interfaces = list(cls.pg_interfaces[0:4])
3971 for i in cls.interfaces:
3976 cls.pg0.generate_remote_hosts(3)
3977 cls.pg0.configure_ipv4_neighbors()
3979 cls.pg1.generate_remote_hosts(1)
3980 cls.pg1.configure_ipv4_neighbors()
3982 cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
3983 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
3984 cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
3986 cls.pg4._local_ip4 = "172.16.255.1"
3987 cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
3988 cls.pg4.set_table_ip4(10)
3989 cls.pg5._local_ip4 = "172.17.255.3"
3990 cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
3991 cls.pg5.set_table_ip4(10)
3992 cls.pg6._local_ip4 = "172.16.255.1"
3993 cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
3994 cls.pg6.set_table_ip4(20)
3995 for i in cls.overlapping_interfaces:
4003 cls.pg9.generate_remote_hosts(2)
4004 cls.pg9.config_ip4()
4005 cls.vapi.sw_interface_add_del_address(
4006 sw_if_index=cls.pg9.sw_if_index,
4007 prefix="10.0.0.1/24")
4010 cls.pg9.resolve_arp()
4011 cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
4012 cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
4013 cls.pg9.resolve_arp()
4016 super(TestNAT44EIMW, self).setUp()
4017 self.vapi.nat44_ei_plugin_enable_disable(
4018 sessions=self.max_translations,
4019 users=self.max_users, enable=1)
4022 super(TestNAT44EIMW, self).tearDown()
4023 if not self.vpp_dead:
4024 self.vapi.nat44_ei_ipfix_enable_disable(
4025 domain_id=self.ipfix_domain_id,
4026 src_port=self.ipfix_src_port,
4028 self.ipfix_src_port = 4739
4029 self.ipfix_domain_id = 1
4031 self.vapi.nat44_ei_plugin_enable_disable(enable=0)
4032 self.vapi.cli("clear logging")
4034 def test_hairpinning(self):
4035 """ NAT44EI hairpinning - 1:1 NAPT """
4037 host = self.pg0.remote_hosts[0]
4038 server = self.pg0.remote_hosts[1]
4041 server_in_port = 5678
4042 server_out_port = 8765
4046 self.nat44_add_address(self.nat_addr)
4047 flags = self.config_flags.NAT44_EI_IF_INSIDE
4048 self.vapi.nat44_ei_interface_add_del_feature(
4049 sw_if_index=self.pg0.sw_if_index,
4050 flags=flags, is_add=1)
4051 self.vapi.nat44_ei_interface_add_del_feature(
4052 sw_if_index=self.pg1.sw_if_index,
4055 # add static mapping for server
4056 self.nat44_add_static_mapping(server.ip4, self.nat_addr,
4057 server_in_port, server_out_port,
4058 proto=IP_PROTOS.tcp)
4060 cnt = self.statistics['/nat44-ei/hairpinning']
4061 # send packet from host to server
4062 p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
4063 IP(src=host.ip4, dst=self.nat_addr) /
4064 TCP(sport=host_in_port, dport=server_out_port))
4065 self.pg0.add_stream(p)
4066 self.pg_enable_capture(self.pg_interfaces)
4068 capture = self.pg0.get_capture(1)
4073 self.assertEqual(ip.src, self.nat_addr)
4074 self.assertEqual(ip.dst, server.ip4)
4075 self.assertNotEqual(tcp.sport, host_in_port)
4076 self.assertEqual(tcp.dport, server_in_port)
4077 self.assert_packet_checksums_valid(p)
4078 host_out_port = tcp.sport
4080 self.logger.error(ppp("Unexpected or invalid packet:", p))
4083 after = self.statistics['/nat44-ei/hairpinning']
4085 if_idx = self.pg0.sw_if_index
4086 self.assertEqual(after[worker_2][if_idx] - cnt[worker_1][if_idx], 1)
4088 # send reply from server to host
4089 p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
4090 IP(src=server.ip4, dst=self.nat_addr) /
4091 TCP(sport=server_in_port, dport=host_out_port))
4092 self.pg0.add_stream(p)
4093 self.pg_enable_capture(self.pg_interfaces)
4095 capture = self.pg0.get_capture(1)
4100 self.assertEqual(ip.src, self.nat_addr)
4101 self.assertEqual(ip.dst, host.ip4)
4102 self.assertEqual(tcp.sport, server_out_port)
4103 self.assertEqual(tcp.dport, host_in_port)
4104 self.assert_packet_checksums_valid(p)
4106 self.logger.error(ppp("Unexpected or invalid packet:", p))
4109 after = self.statistics['/nat44-ei/hairpinning']
4110 if_idx = self.pg0.sw_if_index
4111 self.assertEqual(after[worker_1][if_idx] - cnt[worker_1][if_idx], 1)
4112 self.assertEqual(after[worker_2][if_idx] - cnt[worker_2][if_idx], 2)
4114 def test_hairpinning2(self):
4115 """ NAT44EI hairpinning - 1:1 NAT"""
4117 server1_nat_ip = "10.0.0.10"
4118 server2_nat_ip = "10.0.0.11"
4119 host = self.pg0.remote_hosts[0]
4120 server1 = self.pg0.remote_hosts[1]
4121 server2 = self.pg0.remote_hosts[2]
4122 server_tcp_port = 22
4123 server_udp_port = 20
4125 self.nat44_add_address(self.nat_addr)
4126 flags = self.config_flags.NAT44_EI_IF_INSIDE
4127 self.vapi.nat44_ei_interface_add_del_feature(
4128 sw_if_index=self.pg0.sw_if_index,
4129 flags=flags, is_add=1)
4130 self.vapi.nat44_ei_interface_add_del_feature(
4131 sw_if_index=self.pg1.sw_if_index,
4134 # add static mapping for servers
4135 self.nat44_add_static_mapping(server1.ip4, server1_nat_ip)
4136 self.nat44_add_static_mapping(server2.ip4, server2_nat_ip)
4140 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4141 IP(src=host.ip4, dst=server1_nat_ip) /
4142 TCP(sport=self.tcp_port_in, dport=server_tcp_port))
4144 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4145 IP(src=host.ip4, dst=server1_nat_ip) /
4146 UDP(sport=self.udp_port_in, dport=server_udp_port))
4148 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4149 IP(src=host.ip4, dst=server1_nat_ip) /
4150 ICMP(id=self.icmp_id_in, type='echo-request'))
4152 self.pg0.add_stream(pkts)
4153 self.pg_enable_capture(self.pg_interfaces)
4155 capture = self.pg0.get_capture(len(pkts))
4156 for packet in capture:
4158 self.assertEqual(packet[IP].src, self.nat_addr)
4159 self.assertEqual(packet[IP].dst, server1.ip4)
4160 if packet.haslayer(TCP):
4161 self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
4162 self.assertEqual(packet[TCP].dport, server_tcp_port)
4163 self.tcp_port_out = packet[TCP].sport
4164 self.assert_packet_checksums_valid(packet)
4165 elif packet.haslayer(UDP):
4166 self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
4167 self.assertEqual(packet[UDP].dport, server_udp_port)
4168 self.udp_port_out = packet[UDP].sport
4170 self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
4171 self.icmp_id_out = packet[ICMP].id
4173 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4178 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4179 IP(src=server1.ip4, dst=self.nat_addr) /
4180 TCP(sport=server_tcp_port, dport=self.tcp_port_out))
4182 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4183 IP(src=server1.ip4, dst=self.nat_addr) /
4184 UDP(sport=server_udp_port, dport=self.udp_port_out))
4186 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4187 IP(src=server1.ip4, dst=self.nat_addr) /
4188 ICMP(id=self.icmp_id_out, type='echo-reply'))
4190 self.pg0.add_stream(pkts)
4191 self.pg_enable_capture(self.pg_interfaces)
4193 capture = self.pg0.get_capture(len(pkts))
4194 for packet in capture:
4196 self.assertEqual(packet[IP].src, server1_nat_ip)
4197 self.assertEqual(packet[IP].dst, host.ip4)
4198 if packet.haslayer(TCP):
4199 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
4200 self.assertEqual(packet[TCP].sport, server_tcp_port)
4201 self.assert_packet_checksums_valid(packet)
4202 elif packet.haslayer(UDP):
4203 self.assertEqual(packet[UDP].dport, self.udp_port_in)
4204 self.assertEqual(packet[UDP].sport, server_udp_port)
4206 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4208 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4211 # server2 to server1
4213 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4214 IP(src=server2.ip4, dst=server1_nat_ip) /
4215 TCP(sport=self.tcp_port_in, dport=server_tcp_port))
4217 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4218 IP(src=server2.ip4, dst=server1_nat_ip) /
4219 UDP(sport=self.udp_port_in, dport=server_udp_port))
4221 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4222 IP(src=server2.ip4, dst=server1_nat_ip) /
4223 ICMP(id=self.icmp_id_in, type='echo-request'))
4225 self.pg0.add_stream(pkts)
4226 self.pg_enable_capture(self.pg_interfaces)
4228 capture = self.pg0.get_capture(len(pkts))
4229 for packet in capture:
4231 self.assertEqual(packet[IP].src, server2_nat_ip)
4232 self.assertEqual(packet[IP].dst, server1.ip4)
4233 if packet.haslayer(TCP):
4234 self.assertEqual(packet[TCP].sport, self.tcp_port_in)
4235 self.assertEqual(packet[TCP].dport, server_tcp_port)
4236 self.tcp_port_out = packet[TCP].sport
4237 self.assert_packet_checksums_valid(packet)
4238 elif packet.haslayer(UDP):
4239 self.assertEqual(packet[UDP].sport, self.udp_port_in)
4240 self.assertEqual(packet[UDP].dport, server_udp_port)
4241 self.udp_port_out = packet[UDP].sport
4243 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4244 self.icmp_id_out = packet[ICMP].id
4246 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4249 # server1 to server2
4251 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4252 IP(src=server1.ip4, dst=server2_nat_ip) /
4253 TCP(sport=server_tcp_port, dport=self.tcp_port_out))
4255 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4256 IP(src=server1.ip4, dst=server2_nat_ip) /
4257 UDP(sport=server_udp_port, dport=self.udp_port_out))
4259 p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
4260 IP(src=server1.ip4, dst=server2_nat_ip) /
4261 ICMP(id=self.icmp_id_out, type='echo-reply'))
4263 self.pg0.add_stream(pkts)
4264 self.pg_enable_capture(self.pg_interfaces)
4266 capture = self.pg0.get_capture(len(pkts))
4267 for packet in capture:
4269 self.assertEqual(packet[IP].src, server1_nat_ip)
4270 self.assertEqual(packet[IP].dst, server2.ip4)
4271 if packet.haslayer(TCP):
4272 self.assertEqual(packet[TCP].dport, self.tcp_port_in)
4273 self.assertEqual(packet[TCP].sport, server_tcp_port)
4274 self.assert_packet_checksums_valid(packet)
4275 elif packet.haslayer(UDP):
4276 self.assertEqual(packet[UDP].dport, self.udp_port_in)
4277 self.assertEqual(packet[UDP].sport, server_udp_port)
4279 self.assertEqual(packet[ICMP].id, self.icmp_id_in)
4281 self.logger.error(ppp("Unexpected or invalid packet:", packet))
4285 if __name__ == '__main__':
4286 unittest.main(testRunner=VppTestRunner)