summaryrefslogtreecommitdiffstats
path: root/test/test_snat.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/test_snat.py')
-rw-r--r--test/test_snat.py3718
1 files changed, 0 insertions, 3718 deletions
diff --git a/test/test_snat.py b/test/test_snat.py
deleted file mode 100644
index eb47bbb81d2..00000000000
--- a/test/test_snat.py
+++ /dev/null
@@ -1,3718 +0,0 @@
-#!/usr/bin/env python
-
-import socket
-import unittest
-import struct
-
-from framework import VppTestCase, VppTestRunner, running_extended_tests
-from scapy.layers.inet import IP, TCP, UDP, ICMP
-from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
-from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
-from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6
-from scapy.layers.l2 import Ether, ARP, GRE
-from scapy.data import IP_PROTOS
-from scapy.packet import bind_layers
-from util import ppp
-from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
-from time import sleep
-
-
-class MethodHolder(VppTestCase):
- """ SNAT create capture and verify method holder """
-
- @classmethod
- def setUpClass(cls):
- super(MethodHolder, cls).setUpClass()
-
- def tearDown(self):
- super(MethodHolder, self).tearDown()
-
- def check_ip_checksum(self, pkt):
- """
- Check IP checksum of the packet
-
- :param pkt: Packet to check IP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['IP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['IP'].chksum, pkt['IP'].chksum)
-
- def check_tcp_checksum(self, pkt):
- """
- Check TCP checksum in IP packet
-
- :param pkt: Packet to check TCP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['TCP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum)
-
- def check_udp_checksum(self, pkt):
- """
- Check UDP checksum in IP packet
-
- :param pkt: Packet to check UDP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['UDP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum)
-
- def check_icmp_errror_embedded(self, pkt):
- """
- Check ICMP error embeded packet checksum
-
- :param pkt: Packet to check ICMP error embeded packet checksum
- """
- if pkt.haslayer(IPerror):
- new = pkt.__class__(str(pkt))
- del new['IPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum)
-
- if pkt.haslayer(TCPerror):
- new = pkt.__class__(str(pkt))
- del new['TCPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum)
-
- if pkt.haslayer(UDPerror):
- if pkt['UDPerror'].chksum != 0:
- new = pkt.__class__(str(pkt))
- del new['UDPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['UDPerror'].chksum,
- pkt['UDPerror'].chksum)
-
- if pkt.haslayer(ICMPerror):
- del new['ICMPerror'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum)
-
- def check_icmp_checksum(self, pkt):
- """
- Check ICMP checksum in IPv4 packet
-
- :param pkt: Packet to check ICMP checksum
- """
- new = pkt.__class__(str(pkt))
- del new['ICMP'].chksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum)
- if pkt.haslayer(IPerror):
- self.check_icmp_errror_embedded(pkt)
-
- def check_icmpv6_checksum(self, pkt):
- """
- Check ICMPv6 checksum in IPv4 packet
-
- :param pkt: Packet to check ICMPv6 checksum
- """
- new = pkt.__class__(str(pkt))
- if pkt.haslayer(ICMPv6DestUnreach):
- del new['ICMPv6DestUnreach'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6DestUnreach'].cksum,
- pkt['ICMPv6DestUnreach'].cksum)
- self.check_icmp_errror_embedded(pkt)
- if pkt.haslayer(ICMPv6EchoRequest):
- del new['ICMPv6EchoRequest'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6EchoRequest'].cksum,
- pkt['ICMPv6EchoRequest'].cksum)
- if pkt.haslayer(ICMPv6EchoReply):
- del new['ICMPv6EchoReply'].cksum
- new = new.__class__(str(new))
- self.assertEqual(new['ICMPv6EchoReply'].cksum,
- pkt['ICMPv6EchoReply'].cksum)
-
- def create_stream_in(self, in_if, out_if, ttl=64):
- """
- Create packet stream for inside network
-
- :param in_if: Inside interface
- :param out_if: Outside interface
- :param ttl: TTL of generated packets
- """
- pkts = []
- # TCP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- TCP(sport=self.tcp_port_in, dport=20))
- pkts.append(p)
-
- # UDP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- UDP(sport=self.udp_port_in, dport=20))
- pkts.append(p)
-
- # ICMP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
- pkts.append(p)
-
- return pkts
-
- def compose_ip6(self, ip4, pref, plen):
- """
- Compose IPv4-embedded IPv6 addresses
-
- :param ip4: IPv4 address
- :param pref: IPv6 prefix
- :param plen: IPv6 prefix length
- :returns: IPv4-embedded IPv6 addresses
- """
- pref_n = list(socket.inet_pton(socket.AF_INET6, pref))
- ip4_n = list(socket.inet_pton(socket.AF_INET, ip4))
- if plen == 32:
- pref_n[4] = ip4_n[0]
- pref_n[5] = ip4_n[1]
- pref_n[6] = ip4_n[2]
- pref_n[7] = ip4_n[3]
- elif plen == 40:
- pref_n[5] = ip4_n[0]
- pref_n[6] = ip4_n[1]
- pref_n[7] = ip4_n[2]
- pref_n[9] = ip4_n[3]
- elif plen == 48:
- pref_n[6] = ip4_n[0]
- pref_n[7] = ip4_n[1]
- pref_n[9] = ip4_n[2]
- pref_n[10] = ip4_n[3]
- elif plen == 56:
- pref_n[7] = ip4_n[0]
- pref_n[9] = ip4_n[1]
- pref_n[10] = ip4_n[2]
- pref_n[11] = ip4_n[3]
- elif plen == 64:
- pref_n[9] = ip4_n[0]
- pref_n[10] = ip4_n[1]
- pref_n[11] = ip4_n[2]
- pref_n[12] = ip4_n[3]
- elif plen == 96:
- pref_n[12] = ip4_n[0]
- pref_n[13] = ip4_n[1]
- pref_n[14] = ip4_n[2]
- pref_n[15] = ip4_n[3]
- return socket.inet_ntop(socket.AF_INET6, ''.join(pref_n))
-
- def create_stream_in_ip6(self, in_if, out_if, hlim=64, pref=None, plen=0):
- """
- Create IPv6 packet stream for inside network
-
- :param in_if: Inside interface
- :param out_if: Outside interface
- :param ttl: Hop Limit of generated packets
- :param pref: NAT64 prefix
- :param plen: NAT64 prefix length
- """
- pkts = []
- if pref is None:
- dst = ''.join(['64:ff9b::', out_if.remote_ip4])
- else:
- dst = self.compose_ip6(out_if.remote_ip4, pref, plen)
-
- # TCP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
- TCP(sport=self.tcp_port_in, dport=20))
- pkts.append(p)
-
- # UDP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
- UDP(sport=self.udp_port_in, dport=20))
- pkts.append(p)
-
- # ICMP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IPv6(src=in_if.remote_ip6, dst=dst, hlim=hlim) /
- ICMPv6EchoRequest(id=self.icmp_id_in))
- pkts.append(p)
-
- return pkts
-
- def create_stream_out(self, out_if, dst_ip=None, ttl=64):
- """
- Create packet stream for outside network
-
- :param out_if: Outside interface
- :param dst_ip: Destination IP address (Default use global SNAT address)
- :param ttl: TTL of generated packets
- """
- if dst_ip is None:
- dst_ip = self.snat_addr
- pkts = []
- # TCP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- TCP(dport=self.tcp_port_out, sport=20))
- pkts.append(p)
-
- # UDP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- UDP(dport=self.udp_port_out, sport=20))
- pkts.append(p)
-
- # ICMP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
- pkts.append(p)
-
- return pkts
-
- def verify_capture_out(self, capture, nat_ip=None, same_port=False,
- packet_num=3, dst_ip=None):
- """
- Verify captured packets on outside network
-
- :param capture: Captured packets
- :param nat_ip: Translated IP address (Default use global SNAT address)
- :param same_port: Sorce port number is not translated (Default False)
- :param packet_num: Expected number of packets (Default 3)
- :param dst_ip: Destination IP address (Default do not verify)
- """
- if nat_ip is None:
- nat_ip = self.snat_addr
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.check_ip_checksum(packet)
- self.assertEqual(packet[IP].src, nat_ip)
- if dst_ip is not None:
- self.assertEqual(packet[IP].dst, dst_ip)
- if packet.haslayer(TCP):
- if same_port:
- self.assertEqual(packet[TCP].sport, self.tcp_port_in)
- else:
- self.assertNotEqual(
- packet[TCP].sport, self.tcp_port_in)
- self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- if same_port:
- self.assertEqual(packet[UDP].sport, self.udp_port_in)
- else:
- self.assertNotEqual(
- packet[UDP].sport, self.udp_port_in)
- self.udp_port_out = packet[UDP].sport
- else:
- if same_port:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- else:
- self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
- self.icmp_id_out = packet[ICMP].id
- self.check_icmp_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(outside network):", packet))
- raise
-
- def verify_capture_in(self, capture, in_if, packet_num=3):
- """
- Verify captured packets on inside network
-
- :param capture: Captured packets
- :param in_if: Inside interface
- :param packet_num: Expected number of packets (Default 3)
- """
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.check_ip_checksum(packet)
- self.assertEqual(packet[IP].dst, in_if.remote_ip4)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].dport, self.udp_port_in)
- else:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- self.check_icmp_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(inside network):", packet))
- raise
-
- def verify_capture_in_ip6(self, capture, src_ip, dst_ip, packet_num=3):
- """
- Verify captured IPv6 packets on inside network
-
- :param capture: Captured packets
- :param src_ip: Source IP
- :param dst_ip: Destination IP address
- :param packet_num: Expected number of packets (Default 3)
- """
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, src_ip)
- self.assertEqual(packet[IPv6].dst, dst_ip)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].dport, self.udp_port_in)
- self.check_udp_checksum(packet)
- else:
- self.assertEqual(packet[ICMPv6EchoReply].id,
- self.icmp_id_in)
- self.check_icmpv6_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(inside network):", packet))
- raise
-
- def verify_capture_no_translation(self, capture, ingress_if, egress_if):
- """
- Verify captured packet that don't have to be translated
-
- :param capture: Captured packets
- :param ingress_if: Ingress interface
- :param egress_if: Egress interface
- """
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, ingress_if.remote_ip4)
- self.assertEqual(packet[IP].dst, egress_if.remote_ip4)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].sport, self.tcp_port_in)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].sport, self.udp_port_in)
- else:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(inside network):", packet))
- raise
-
- def verify_capture_out_with_icmp_errors(self, capture, src_ip=None,
- packet_num=3, icmp_type=11):
- """
- Verify captured packets with ICMP errors on outside network
-
- :param capture: Captured packets
- :param src_ip: Translated IP address or IP address of VPP
- (Default use global SNAT address)
- :param packet_num: Expected number of packets (Default 3)
- :param icmp_type: Type of error ICMP packet
- we are expecting (Default 11)
- """
- if src_ip is None:
- src_ip = self.snat_addr
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, src_ip)
- self.assertTrue(packet.haslayer(ICMP))
- icmp = packet[ICMP]
- self.assertEqual(icmp.type, icmp_type)
- self.assertTrue(icmp.haslayer(IPerror))
- inner_ip = icmp[IPerror]
- if inner_ip.haslayer(TCPerror):
- self.assertEqual(inner_ip[TCPerror].dport,
- self.tcp_port_out)
- elif inner_ip.haslayer(UDPerror):
- self.assertEqual(inner_ip[UDPerror].dport,
- self.udp_port_out)
- else:
- self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_out)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(outside network):", packet))
- raise
-
- def verify_capture_in_with_icmp_errors(self, capture, in_if, packet_num=3,
- icmp_type=11):
- """
- Verify captured packets with ICMP errors on inside network
-
- :param capture: Captured packets
- :param in_if: Inside interface
- :param packet_num: Expected number of packets (Default 3)
- :param icmp_type: Type of error ICMP packet
- we are expecting (Default 11)
- """
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].dst, in_if.remote_ip4)
- self.assertTrue(packet.haslayer(ICMP))
- icmp = packet[ICMP]
- self.assertEqual(icmp.type, icmp_type)
- self.assertTrue(icmp.haslayer(IPerror))
- inner_ip = icmp[IPerror]
- if inner_ip.haslayer(TCPerror):
- self.assertEqual(inner_ip[TCPerror].sport,
- self.tcp_port_in)
- elif inner_ip.haslayer(UDPerror):
- self.assertEqual(inner_ip[UDPerror].sport,
- self.udp_port_in)
- else:
- self.assertEqual(inner_ip[ICMPerror].id, self.icmp_id_in)
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(inside network):", packet))
- raise
-
- def verify_ipfix_nat44_ses(self, data):
- """
- Verify IPFIX NAT44 session create/delete event
-
- :param data: Decoded IPFIX data records
- """
- nat44_ses_create_num = 0
- nat44_ses_delete_num = 0
- self.assertEqual(6, len(data))
- for record in data:
- # natEvent
- self.assertIn(ord(record[230]), [4, 5])
- if ord(record[230]) == 4:
- nat44_ses_create_num += 1
- else:
- nat44_ses_delete_num += 1
- # sourceIPv4Address
- self.assertEqual(self.pg0.remote_ip4n, record[8])
- # postNATSourceIPv4Address
- self.assertEqual(socket.inet_pton(socket.AF_INET, self.snat_addr),
- record[225])
- # ingressVRFID
- self.assertEqual(struct.pack("!I", 0), record[234])
- # protocolIdentifier/sourceTransportPort/postNAPTSourceTransportPort
- if IP_PROTOS.icmp == ord(record[4]):
- self.assertEqual(struct.pack("!H", self.icmp_id_in), record[7])
- self.assertEqual(struct.pack("!H", self.icmp_id_out),
- record[227])
- elif IP_PROTOS.tcp == ord(record[4]):
- self.assertEqual(struct.pack("!H", self.tcp_port_in),
- record[7])
- self.assertEqual(struct.pack("!H", self.tcp_port_out),
- record[227])
- elif IP_PROTOS.udp == ord(record[4]):
- self.assertEqual(struct.pack("!H", self.udp_port_in),
- record[7])
- self.assertEqual(struct.pack("!H", self.udp_port_out),
- record[227])
- else:
- self.fail("Invalid protocol")
- self.assertEqual(3, nat44_ses_create_num)
- self.assertEqual(3, nat44_ses_delete_num)
-
- def verify_ipfix_addr_exhausted(self, data):
- """
- Verify IPFIX NAT addresses event
-
- :param data: Decoded IPFIX data records
- """
- self.assertEqual(1, len(data))
- record = data[0]
- # natEvent
- self.assertEqual(ord(record[230]), 3)
- # natPoolID
- self.assertEqual(struct.pack("!I", 0), record[283])
-
-
-class TestSNAT(MethodHolder):
- """ SNAT Test Cases """
-
- @classmethod
- def setUpClass(cls):
- super(TestSNAT, cls).setUpClass()
-
- try:
- cls.tcp_port_in = 6303
- cls.tcp_port_out = 6303
- cls.udp_port_in = 6304
- cls.udp_port_out = 6304
- cls.icmp_id_in = 6305
- cls.icmp_id_out = 6305
- cls.snat_addr = '10.0.0.3'
- cls.ipfix_src_port = 4739
- cls.ipfix_domain_id = 1
-
- cls.create_pg_interfaces(range(9))
- cls.interfaces = list(cls.pg_interfaces[0:4])
-
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- cls.pg0.generate_remote_hosts(3)
- cls.pg0.configure_ipv4_neighbors()
-
- cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
-
- cls.pg4._local_ip4 = "172.16.255.1"
- cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
- cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
- cls.pg4.set_table_ip4(10)
- cls.pg5._local_ip4 = "172.17.255.3"
- cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
- cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
- cls.pg5.set_table_ip4(10)
- cls.pg6._local_ip4 = "172.16.255.1"
- cls.pg6._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
- cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
- cls.pg6.set_table_ip4(20)
- for i in cls.overlapping_interfaces:
- i.config_ip4()
- i.admin_up()
- i.resolve_arp()
-
- cls.pg7.admin_up()
- cls.pg8.admin_up()
-
- except Exception:
- super(TestSNAT, cls).tearDownClass()
- raise
-
- def clear_snat(self):
- """
- Clear SNAT configuration.
- """
- # I found no elegant way to do this
- self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg7.remote_ip4n,
- next_hop_sw_if_index=self.pg7.sw_if_index,
- is_add=0)
- self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg8.remote_ip4n,
- next_hop_sw_if_index=self.pg8.sw_if_index,
- is_add=0)
-
- for intf in [self.pg7, self.pg8]:
- neighbors = self.vapi.ip_neighbor_dump(intf.sw_if_index)
- for n in neighbors:
- self.vapi.ip_neighbor_add_del(intf.sw_if_index,
- n.mac_address,
- n.ip_address,
- is_add=0)
-
- if self.pg7.has_ip4_config:
- self.pg7.unconfig_ip4()
-
- interfaces = self.vapi.snat_interface_addr_dump()
- for intf in interfaces:
- self.vapi.snat_add_interface_addr(intf.sw_if_index, is_add=0)
-
- self.vapi.snat_ipfix(enable=0, src_port=self.ipfix_src_port,
- domain_id=self.ipfix_domain_id)
- self.ipfix_src_port = 4739
- self.ipfix_domain_id = 1
-
- interfaces = self.vapi.snat_interface_dump()
- for intf in interfaces:
- self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
-
- interfaces = self.vapi.snat_interface_output_feature_dump()
- for intf in interfaces:
- self.vapi.snat_interface_add_del_output_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
-
- static_mappings = self.vapi.snat_static_mapping_dump()
- for sm in static_mappings:
- self.vapi.snat_add_static_mapping(sm.local_ip_address,
- sm.external_ip_address,
- local_port=sm.local_port,
- external_port=sm.external_port,
- addr_only=sm.addr_only,
- vrf_id=sm.vrf_id,
- protocol=sm.protocol,
- is_add=0)
-
- adresses = self.vapi.snat_address_dump()
- for addr in adresses:
- self.vapi.snat_add_address_range(addr.ip_address,
- addr.ip_address,
- is_add=0)
-
- def snat_add_static_mapping(self, local_ip, external_ip='0.0.0.0',
- local_port=0, external_port=0, vrf_id=0,
- is_add=1, external_sw_if_index=0xFFFFFFFF,
- proto=0):
- """
- Add/delete S-NAT static mapping
-
- :param local_ip: Local IP address
- :param external_ip: External IP address
- :param local_port: Local port number (Optional)
- :param external_port: External port number (Optional)
- :param vrf_id: VRF ID (Default 0)
- :param is_add: 1 if add, 0 if delete (Default add)
- :param external_sw_if_index: External interface instead of IP address
- :param proto: IP protocol (Mandatory if port specified)
- """
- addr_only = 1
- if local_port and external_port:
- addr_only = 0
- l_ip = socket.inet_pton(socket.AF_INET, local_ip)
- e_ip = socket.inet_pton(socket.AF_INET, external_ip)
- self.vapi.snat_add_static_mapping(
- l_ip,
- e_ip,
- external_sw_if_index,
- local_port,
- external_port,
- addr_only,
- vrf_id,
- proto,
- is_add)
-
- def snat_add_address(self, ip, is_add=1, vrf_id=0xFFFFFFFF):
- """
- Add/delete S-NAT address
-
- :param ip: IP address
- :param is_add: 1 if add, 0 if delete (Default add)
- """
- snat_addr = socket.inet_pton(socket.AF_INET, ip)
- self.vapi.snat_add_address_range(snat_addr, snat_addr, is_add,
- vrf_id=vrf_id)
-
- def test_dynamic(self):
- """ SNAT dynamic translation test """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- def test_dynamic_icmp_errors_in2out_ttl_1(self):
- """ SNAT handling of client packets with TTL=1 """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # Client side - generate traffic
- pkts = self.create_stream_in(self.pg0, self.pg1, ttl=1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Client side - verify ICMP type 11 packets
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in_with_icmp_errors(capture, self.pg0)
-
- def test_dynamic_icmp_errors_out2in_ttl_1(self):
- """ SNAT handling of server packets with TTL=1 """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # Client side - create sessions
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Server side - generate traffic
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
- pkts = self.create_stream_out(self.pg1, ttl=1)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Server side - verify ICMP type 11 packets
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out_with_icmp_errors(capture,
- src_ip=self.pg1.local_ip4)
-
- def test_dynamic_icmp_errors_in2out_ttl_2(self):
- """ SNAT handling of error responses to client packets with TTL=2 """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # Client side - generate traffic
- pkts = self.create_stream_in(self.pg0, self.pg1, ttl=2)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Server side - simulate ICMP type 11 response
- capture = self.pg1.get_capture(len(pkts))
- pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- ICMP(type=11) / packet[IP] for packet in capture]
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Client side - verify ICMP type 11 packets
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in_with_icmp_errors(capture, self.pg0)
-
- def test_dynamic_icmp_errors_out2in_ttl_2(self):
- """ SNAT handling of error responses to server packets with TTL=2 """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # Client side - create sessions
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Server side - generate traffic
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
- pkts = self.create_stream_out(self.pg1, ttl=2)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Client side - simulate ICMP type 11 response
- capture = self.pg0.get_capture(len(pkts))
- pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- ICMP(type=11) / packet[IP] for packet in capture]
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # Server side - verify ICMP type 11 packets
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out_with_icmp_errors(capture)
-
- def test_ping_out_interface_from_outside(self):
- """ Ping SNAT out interface from outside network """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.pg1.local_ip4) /
- ICMP(id=self.icmp_id_out, type='echo-request'))
- pkts = [p]
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.assertEqual(1, len(capture))
- packet = capture[0]
- try:
- self.assertEqual(packet[IP].src, self.pg1.local_ip4)
- self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- self.assertEqual(packet[ICMP].type, 0) # echo reply
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(outside network):", packet))
- raise
-
- def test_ping_internal_host_from_outside(self):
- """ Ping internal host from outside network """
-
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # out2in
- pkt = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr, ttl=64) /
- ICMP(id=self.icmp_id_out, type='echo-request'))
- self.pg1.add_stream(pkt)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- self.verify_capture_in(capture, self.pg0, packet_num=1)
- self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
-
- # in2out
- pkt = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
- ICMP(id=self.icmp_id_in, type='echo-reply'))
- self.pg0.add_stream(pkt)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- self.verify_capture_out(capture, same_port=True, packet_num=1)
- self.assert_equal(capture[0][IP].proto, IP_PROTOS.icmp)
-
- def test_static_in(self):
- """ SNAT 1:1 NAT initialized from inside network """
-
- nat_ip = "10.0.0.10"
- self.tcp_port_out = 6303
- self.udp_port_out = 6304
- self.icmp_id_out = 6305
-
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip, True)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, nat_ip)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- def test_static_out(self):
- """ SNAT 1:1 NAT initialized from outside network """
-
- nat_ip = "10.0.0.20"
- self.tcp_port_out = 6303
- self.udp_port_out = 6304
- self.icmp_id_out = 6305
-
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, nat_ip)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip, True)
-
- def test_static_with_port_in(self):
- """ SNAT 1:1 NAT with port initialized from inside network """
-
- self.tcp_port_out = 3606
- self.udp_port_out = 3607
- self.icmp_id_out = 3608
-
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- def test_static_with_port_out(self):
- """ SNAT 1:1 NAT with port initialized from outside network """
-
- self.tcp_port_out = 30606
- self.udp_port_out = 30607
- self.icmp_id_out = 30608
-
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg0.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- def test_static_vrf_aware(self):
- """ SNAT 1:1 NAT VRF awareness """
-
- nat_ip1 = "10.0.0.30"
- nat_ip2 = "10.0.0.40"
- self.tcp_port_out = 6303
- self.udp_port_out = 6304
- self.icmp_id_out = 6305
-
- self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
- vrf_id=10)
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
- vrf_id=10)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
-
- # inside interface VRF match SNAT static mapping VRF
- pkts = self.create_stream_in(self.pg4, self.pg3)
- self.pg4.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip1, True)
-
- # inside interface VRF don't match SNAT static mapping VRF (packets
- # are dropped)
- pkts = self.create_stream_in(self.pg0, self.pg3)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg3.assert_nothing_captured()
-
- def test_multiple_inside_interfaces(self):
- """ SNAT multiple inside interfaces (non-overlapping address space) """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
-
- # between two S-NAT inside interfaces (no translation)
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_no_translation(capture, self.pg0, self.pg1)
-
- # from S-NAT inside to interface without S-NAT feature (no translation)
- pkts = self.create_stream_in(self.pg0, self.pg2)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_no_translation(capture, self.pg0, self.pg2)
-
- # in2out 1st interface
- pkts = self.create_stream_in(self.pg0, self.pg3)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in 1st interface
- pkts = self.create_stream_out(self.pg3)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- # in2out 2nd interface
- pkts = self.create_stream_in(self.pg1, self.pg3)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in 2nd interface
- pkts = self.create_stream_out(self.pg3)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg1)
-
- def test_inside_overlapping_interfaces(self):
- """ SNAT multiple inside interfaces with overlapping address space """
-
- static_nat_ip = "10.0.0.10"
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg3.sw_if_index,
- is_inside=0)
- self.vapi.snat_interface_add_del_feature(self.pg4.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg5.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg6.sw_if_index)
- self.snat_add_static_mapping(self.pg6.remote_ip4, static_nat_ip,
- vrf_id=20)
-
- # between S-NAT inside interfaces with same VRF (no translation)
- pkts = self.create_stream_in(self.pg4, self.pg5)
- self.pg4.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg5.get_capture(len(pkts))
- self.verify_capture_no_translation(capture, self.pg4, self.pg5)
-
- # between S-NAT inside interfaces with different VRF (hairpinning)
- p = (Ether(src=self.pg4.remote_mac, dst=self.pg4.local_mac) /
- IP(src=self.pg4.remote_ip4, dst=static_nat_ip) /
- TCP(sport=1234, dport=5678))
- self.pg4.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg6.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
- self.assertEqual(ip.dst, self.pg6.remote_ip4)
- self.assertNotEqual(tcp.sport, 1234)
- self.assertEqual(tcp.dport, 5678)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- # in2out 1st interface
- pkts = self.create_stream_in(self.pg4, self.pg3)
- self.pg4.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in 1st interface
- pkts = self.create_stream_out(self.pg3)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg4.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg4)
-
- # in2out 2nd interface
- pkts = self.create_stream_in(self.pg5, self.pg3)
- self.pg5.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in 2nd interface
- pkts = self.create_stream_out(self.pg3)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg5.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg5)
-
- # pg5 session dump
- addresses = self.vapi.snat_address_dump()
- self.assertEqual(len(addresses), 1)
- sessions = self.vapi.snat_user_session_dump(self.pg5.remote_ip4n, 10)
- self.assertEqual(len(sessions), 3)
- for session in sessions:
- self.assertFalse(session.is_static)
- self.assertEqual(session.inside_ip_address[0:4],
- self.pg5.remote_ip4n)
- self.assertEqual(session.outside_ip_address,
- addresses[0].ip_address)
- self.assertEqual(sessions[0].protocol, IP_PROTOS.tcp)
- self.assertEqual(sessions[1].protocol, IP_PROTOS.udp)
- self.assertEqual(sessions[2].protocol, IP_PROTOS.icmp)
- self.assertEqual(sessions[0].inside_port, self.tcp_port_in)
- self.assertEqual(sessions[1].inside_port, self.udp_port_in)
- self.assertEqual(sessions[2].inside_port, self.icmp_id_in)
- self.assertEqual(sessions[0].outside_port, self.tcp_port_out)
- self.assertEqual(sessions[1].outside_port, self.udp_port_out)
- self.assertEqual(sessions[2].outside_port, self.icmp_id_out)
-
- # in2out 3rd interface
- pkts = self.create_stream_in(self.pg6, self.pg3)
- self.pg6.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture, static_nat_ip, True)
-
- # out2in 3rd interface
- pkts = self.create_stream_out(self.pg3, static_nat_ip)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg6.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg6)
-
- # general user and session dump verifications
- users = self.vapi.snat_user_dump()
- self.assertTrue(len(users) >= 3)
- addresses = self.vapi.snat_address_dump()
- self.assertEqual(len(addresses), 1)
- for user in users:
- sessions = self.vapi.snat_user_session_dump(user.ip_address,
- user.vrf_id)
- for session in sessions:
- self.assertEqual(user.ip_address, session.inside_ip_address)
- self.assertTrue(session.total_bytes > session.total_pkts > 0)
- self.assertTrue(session.protocol in
- [IP_PROTOS.tcp, IP_PROTOS.udp,
- IP_PROTOS.icmp])
-
- # pg4 session dump
- sessions = self.vapi.snat_user_session_dump(self.pg4.remote_ip4n, 10)
- self.assertTrue(len(sessions) >= 4)
- for session in sessions:
- self.assertFalse(session.is_static)
- self.assertEqual(session.inside_ip_address[0:4],
- self.pg4.remote_ip4n)
- self.assertEqual(session.outside_ip_address,
- addresses[0].ip_address)
-
- # pg6 session dump
- sessions = self.vapi.snat_user_session_dump(self.pg6.remote_ip4n, 20)
- self.assertTrue(len(sessions) >= 3)
- for session in sessions:
- self.assertTrue(session.is_static)
- self.assertEqual(session.inside_ip_address[0:4],
- self.pg6.remote_ip4n)
- self.assertEqual(map(ord, session.outside_ip_address[0:4]),
- map(int, static_nat_ip.split('.')))
- self.assertTrue(session.inside_port in
- [self.tcp_port_in, self.udp_port_in,
- self.icmp_id_in])
-
- def test_hairpinning(self):
- """ SNAT hairpinning - 1:1 NAT with port"""
-
- host = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
- host_in_port = 1234
- host_out_port = 0
- server_in_port = 5678
- server_out_port = 8765
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- # add static mapping for server
- self.snat_add_static_mapping(server.ip4, self.snat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
-
- # send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.snat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
- self.assertEqual(ip.dst, server.ip4)
- self.assertNotEqual(tcp.sport, host_in_port)
- self.assertEqual(tcp.dport, server_in_port)
- self.check_tcp_checksum(p)
- host_out_port = tcp.sport
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- # send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
- self.assertEqual(ip.dst, host.ip4)
- self.assertEqual(tcp.sport, server_out_port)
- self.assertEqual(tcp.dport, host_in_port)
- self.check_tcp_checksum(p)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:"), p)
- raise
-
- def test_hairpinning2(self):
- """ SNAT hairpinning - 1:1 NAT"""
-
- server1_nat_ip = "10.0.0.10"
- server2_nat_ip = "10.0.0.11"
- host = self.pg0.remote_hosts[0]
- server1 = self.pg0.remote_hosts[1]
- server2 = self.pg0.remote_hosts[2]
- server_tcp_port = 22
- server_udp_port = 20
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # add static mapping for servers
- self.snat_add_static_mapping(server1.ip4, server1_nat_ip)
- self.snat_add_static_mapping(server2.ip4, server2_nat_ip)
-
- # host to server1
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=host.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, self.snat_addr)
- self.assertEqual(packet[IP].dst, server1.ip4)
- if packet.haslayer(TCP):
- self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
- self.assertEqual(packet[TCP].dport, server_tcp_port)
- self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertNotEqual(packet[UDP].sport, self.udp_port_in)
- self.assertEqual(packet[UDP].dport, server_udp_port)
- self.udp_port_out = packet[UDP].sport
- else:
- self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
- self.icmp_id_out = packet[ICMP].id
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server1 to host
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=self.snat_addr) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, server1_nat_ip)
- self.assertEqual(packet[IP].dst, host.ip4)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.assertEqual(packet[TCP].sport, server_tcp_port)
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].dport, self.udp_port_in)
- self.assertEqual(packet[UDP].sport, server_udp_port)
- else:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server2 to server1
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- TCP(sport=self.tcp_port_in, dport=server_tcp_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- UDP(sport=self.udp_port_in, dport=server_udp_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server2.ip4, dst=server1_nat_ip) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, server2_nat_ip)
- self.assertEqual(packet[IP].dst, server1.ip4)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].sport, self.tcp_port_in)
- self.assertEqual(packet[TCP].dport, server_tcp_port)
- self.tcp_port_out = packet[TCP].sport
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].sport, self.udp_port_in)
- self.assertEqual(packet[UDP].dport, server_udp_port)
- self.udp_port_out = packet[UDP].sport
- else:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- self.icmp_id_out = packet[ICMP].id
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server1 to server2
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- TCP(sport=server_tcp_port, dport=self.tcp_port_out))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- UDP(sport=server_udp_port, dport=self.udp_port_out))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=server1.ip4, dst=server2_nat_ip) /
- ICMP(id=self.icmp_id_out, type='echo-reply'))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, server1_nat_ip)
- self.assertEqual(packet[IP].dst, server2.ip4)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].dport, self.tcp_port_in)
- self.assertEqual(packet[TCP].sport, server_tcp_port)
- self.check_tcp_checksum(packet)
- elif packet.haslayer(UDP):
- self.assertEqual(packet[UDP].dport, self.udp_port_in)
- self.assertEqual(packet[UDP].sport, server_udp_port)
- else:
- self.assertEqual(packet[ICMP].id, self.icmp_id_in)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_max_translations_per_user(self):
- """ MAX translations per user - recycle the least recently used """
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # get maximum number of translations per user
- snat_config = self.vapi.snat_show_config()
-
- # send more than maximum number of translations per user packets
- pkts_num = snat_config.max_translations_per_user + 5
- pkts = []
- for port in range(0, pkts_num):
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=1025 + port))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
-
- # verify number of translated packet
- self.pg1.get_capture(pkts_num)
-
- def test_interface_addr(self):
- """ Acquire SNAT addresses from interface """
- self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
-
- # no address in NAT pool
- adresses = self.vapi.snat_address_dump()
- self.assertEqual(0, len(adresses))
-
- # configure interface address and check NAT address pool
- self.pg7.config_ip4()
- adresses = self.vapi.snat_address_dump()
- self.assertEqual(1, len(adresses))
- self.assertEqual(adresses[0].ip_address[0:4], self.pg7.local_ip4n)
-
- # remove interface address and check NAT address pool
- self.pg7.unconfig_ip4()
- adresses = self.vapi.snat_address_dump()
- self.assertEqual(0, len(adresses))
-
- def test_interface_addr_static_mapping(self):
- """ Static mapping with addresses from interface """
- self.vapi.snat_add_interface_addr(self.pg7.sw_if_index)
- self.snat_add_static_mapping('1.2.3.4',
- external_sw_if_index=self.pg7.sw_if_index)
-
- # static mappings with external interface
- static_mappings = self.vapi.snat_static_mapping_dump()
- self.assertEqual(1, len(static_mappings))
- self.assertEqual(self.pg7.sw_if_index,
- static_mappings[0].external_sw_if_index)
-
- # configure interface address and check static mappings
- self.pg7.config_ip4()
- static_mappings = self.vapi.snat_static_mapping_dump()
- self.assertEqual(1, len(static_mappings))
- self.assertEqual(static_mappings[0].external_ip_address[0:4],
- self.pg7.local_ip4n)
- self.assertEqual(0xFFFFFFFF, static_mappings[0].external_sw_if_index)
-
- # remove interface address and check static mappings
- self.pg7.unconfig_ip4()
- static_mappings = self.vapi.snat_static_mapping_dump()
- self.assertEqual(0, len(static_mappings))
-
- def test_ipfix_nat44_sess(self):
- """ S-NAT IPFIX logging NAT44 session created/delted """
- self.ipfix_domain_id = 10
- self.ipfix_src_port = 20202
- colector_port = 30303
- bind_layers(UDP, IPFIX, dport=30303)
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
- src_address=self.pg3.local_ip4n,
- path_mtu=512,
- template_interval=10,
- collector_port=colector_port)
- self.vapi.snat_ipfix(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port)
-
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
- self.snat_add_address(self.snat_addr, is_add=0)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
- capture = self.pg3.get_capture(3)
- ipfix = IPFIXDecoder()
- # first load template
- for p in capture:
- self.assertTrue(p.haslayer(IPFIX))
- self.assertEqual(p[IP].src, self.pg3.local_ip4)
- self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
- self.assertEqual(p[UDP].sport, self.ipfix_src_port)
- self.assertEqual(p[UDP].dport, colector_port)
- self.assertEqual(p[IPFIX].observationDomainID,
- self.ipfix_domain_id)
- if p.haslayer(Template):
- ipfix.add_template(p.getlayer(Template))
- # verify events in data set
- for p in capture:
- if p.haslayer(Data):
- data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_nat44_ses(data)
-
- def test_ipfix_addr_exhausted(self):
- """ S-NAT IPFIX logging NAT addresses exhausted """
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- self.vapi.set_ipfix_exporter(collector_address=self.pg3.remote_ip4n,
- src_address=self.pg3.local_ip4n,
- path_mtu=512,
- template_interval=10)
- self.vapi.snat_ipfix(domain_id=self.ipfix_domain_id,
- src_port=self.ipfix_src_port)
-
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=3025))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(0)
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
- capture = self.pg3.get_capture(3)
- ipfix = IPFIXDecoder()
- # first load template
- for p in capture:
- self.assertTrue(p.haslayer(IPFIX))
- self.assertEqual(p[IP].src, self.pg3.local_ip4)
- self.assertEqual(p[IP].dst, self.pg3.remote_ip4)
- self.assertEqual(p[UDP].sport, self.ipfix_src_port)
- self.assertEqual(p[UDP].dport, 4739)
- self.assertEqual(p[IPFIX].observationDomainID,
- self.ipfix_domain_id)
- if p.haslayer(Template):
- ipfix.add_template(p.getlayer(Template))
- # verify events in data set
- for p in capture:
- if p.haslayer(Data):
- data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_addr_exhausted(data)
-
- def test_pool_addr_fib(self):
- """ S-NAT add pool addresses to FIB """
- static_addr = '10.0.0.10'
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr)
-
- # SNAT address
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- self.assertTrue(capture[0].haslayer(ARP))
- self.assertTrue(capture[0][ARP].op, ARP.is_at)
-
- # 1:1 NAT address
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=static_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- self.assertTrue(capture[0].haslayer(ARP))
- self.assertTrue(capture[0][ARP].op, ARP.is_at)
-
- # send ARP to non-SNAT interface
- p = (Ether(src=self.pg2.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
- psrc=self.pg2.remote_ip4, hwsrc=self.pg2.remote_mac))
- self.pg2.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(0)
-
- # remove addresses and verify
- self.snat_add_address(self.snat_addr, is_add=0)
- self.snat_add_static_mapping(self.pg0.remote_ip4, static_addr,
- is_add=0)
-
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=self.snat_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(0)
-
- p = (Ether(src=self.pg1.remote_mac, dst='ff:ff:ff:ff:ff:ff') /
- ARP(op=ARP.who_has, pdst=static_addr,
- psrc=self.pg1.remote_ip4, hwsrc=self.pg1.remote_mac))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(0)
-
- def test_vrf_mode(self):
- """ S-NAT tenant VRF aware address pool mode """
-
- vrf_id1 = 1
- vrf_id2 = 2
- nat_ip1 = "10.0.0.10"
- nat_ip2 = "10.0.0.11"
-
- self.pg0.unconfig_ip4()
- self.pg1.unconfig_ip4()
- self.pg0.set_table_ip4(vrf_id1)
- self.pg1.set_table_ip4(vrf_id2)
- self.pg0.config_ip4()
- self.pg1.config_ip4()
-
- self.snat_add_address(nat_ip1, vrf_id=vrf_id1)
- self.snat_add_address(nat_ip2, vrf_id=vrf_id2)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
- is_inside=0)
-
- # first VRF
- pkts = self.create_stream_in(self.pg0, self.pg2)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip1)
-
- # second VRF
- pkts = self.create_stream_in(self.pg1, self.pg2)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip2)
-
- def test_vrf_feature_independent(self):
- """ S-NAT tenant VRF independent address pool mode """
-
- nat_ip1 = "10.0.0.10"
- nat_ip2 = "10.0.0.11"
-
- self.snat_add_address(nat_ip1)
- self.snat_add_address(nat_ip2)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg2.sw_if_index,
- is_inside=0)
-
- # first VRF
- pkts = self.create_stream_in(self.pg0, self.pg2)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip1)
-
- # second VRF
- pkts = self.create_stream_in(self.pg1, self.pg2)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip1)
-
- def test_dynamic_ipless_interfaces(self):
- """ SNAT interfaces without configured ip dynamic map """
-
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- self.pg8.remote_mac,
- self.pg8.remote_ip4n,
- is_static=1)
-
- self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg7.remote_ip4n,
- next_hop_sw_if_index=self.pg7.sw_if_index)
- self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg8.remote_ip4n,
- next_hop_sw_if_index=self.pg8.sw_if_index)
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg7, self.pg8)
- self.pg7.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg8.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in
- pkts = self.create_stream_out(self.pg8, self.snat_addr)
- self.pg8.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg7.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg7)
-
- def test_static_ipless_interfaces(self):
- """ SNAT 1:1 NAT interfaces without configured ip """
-
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- self.pg8.remote_mac,
- self.pg8.remote_ip4n,
- is_static=1)
-
- self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg7.remote_ip4n,
- next_hop_sw_if_index=self.pg7.sw_if_index)
- self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg8.remote_ip4n,
- next_hop_sw_if_index=self.pg8.sw_if_index)
-
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
-
- # out2in
- pkts = self.create_stream_out(self.pg8)
- self.pg8.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg7.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg7)
-
- # in2out
- pkts = self.create_stream_in(self.pg7, self.pg8)
- self.pg7.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg8.get_capture(len(pkts))
- self.verify_capture_out(capture, self.snat_addr, True)
-
- def test_static_with_port_ipless_interfaces(self):
- """ SNAT 1:1 NAT with port interfaces without configured ip """
-
- self.tcp_port_out = 30606
- self.udp_port_out = 30607
- self.icmp_id_out = 30608
-
- self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
- self.pg7.remote_ip4n,
- is_static=1)
- self.vapi.ip_neighbor_add_del(self.pg8.sw_if_index,
- self.pg8.remote_mac,
- self.pg8.remote_ip4n,
- is_static=1)
-
- self.vapi.ip_add_del_route(dst_address=self.pg7.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg7.remote_ip4n,
- next_hop_sw_if_index=self.pg7.sw_if_index)
- self.vapi.ip_add_del_route(dst_address=self.pg8.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg8.remote_ip4n,
- next_hop_sw_if_index=self.pg8.sw_if_index)
-
- self.snat_add_address(self.snat_addr)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.tcp_port_in, self.tcp_port_out,
- proto=IP_PROTOS.tcp)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.udp_port_in, self.udp_port_out,
- proto=IP_PROTOS.udp)
- self.snat_add_static_mapping(self.pg7.remote_ip4, self.snat_addr,
- self.icmp_id_in, self.icmp_id_out,
- proto=IP_PROTOS.icmp)
- self.vapi.snat_interface_add_del_feature(self.pg7.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg8.sw_if_index,
- is_inside=0)
-
- # out2in
- pkts = self.create_stream_out(self.pg8)
- self.pg8.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg7.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg7)
-
- # in2out
- pkts = self.create_stream_in(self.pg7, self.pg8)
- self.pg7.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg8.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- def test_static_unknown_proto(self):
- """ 1:1 NAT translate packet with unknown protocol """
- nat_ip = "10.0.0.10"
- self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg1.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, nat_ip)
- self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # out2in
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=nat_ip) /
- GRE() /
- IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
- self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_hairpinning_static_unknown_proto(self):
- """ 1:1 NAT translate packet with unknown protocol - hairpinning """
-
- host = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
-
- host_nat_ip = "10.0.0.10"
- server_nat_ip = "10.0.0.11"
-
- self.snat_add_static_mapping(host.ip4, host_nat_ip)
- self.snat_add_static_mapping(server.ip4, server_nat_ip)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # host to server
- p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
- IP(src=host.ip4, dst=server_nat_ip) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, host_nat_ip)
- self.assertEqual(packet[IP].dst, server.ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server to host
- p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
- IP(src=server.ip4, dst=host_nat_ip) /
- GRE() /
- IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, server_nat_ip)
- self.assertEqual(packet[IP].dst, host.ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_unknown_proto(self):
- """ SNAT translate packet with unknown protocol """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=20))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg1.get_capture(1)
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg1.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, self.snat_addr)
- self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # out2in
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- GRE() /
- IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, self.pg1.remote_ip4)
- self.assertEqual(packet[IP].dst, self.pg0.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_hairpinning_unknown_proto(self):
- """ SNAT translate packet with unknown protocol - hairpinning """
- host = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
- host_in_port = 1234
- host_out_port = 0
- server_in_port = 5678
- server_out_port = 8765
- server_nat_ip = "10.0.0.11"
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # add static mapping for server
- self.snat_add_static_mapping(server.ip4, server_nat_ip)
-
- # host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=server_nat_ip) /
- TCP(sport=host_in_port, dport=server_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
-
- p = (Ether(dst=self.pg0.local_mac, src=host.mac) /
- IP(src=host.ip4, dst=server_nat_ip) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg3.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, self.snat_addr)
- self.assertEqual(packet[IP].dst, server.ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server to host
- p = (Ether(dst=self.pg0.local_mac, src=server.mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
- GRE() /
- IP(src=self.pg3.remote_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, server_nat_ip)
- self.assertEqual(packet[IP].dst, host.ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_output_feature(self):
- """ S-NAT interface output feature (in2out postrouting) """
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_output_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
-
- # out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- def test_output_feature_vrf_aware(self):
- """ S-NAT interface output feature VRF aware (in2out postrouting) """
- nat_ip_vrf10 = "10.0.0.10"
- nat_ip_vrf20 = "10.0.0.20"
-
- self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg3.remote_ip4n,
- next_hop_sw_if_index=self.pg3.sw_if_index,
- table_id=10)
- self.vapi.ip_add_del_route(dst_address=self.pg3.remote_ip4n,
- dst_address_length=32,
- next_hop_address=self.pg3.remote_ip4n,
- next_hop_sw_if_index=self.pg3.sw_if_index,
- table_id=20)
-
- self.snat_add_address(nat_ip_vrf10, vrf_id=10)
- self.snat_add_address(nat_ip_vrf20, vrf_id=20)
- self.vapi.snat_interface_add_del_output_feature(self.pg4.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg6.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg3.sw_if_index,
- is_inside=0)
-
- # in2out VRF 10
- pkts = self.create_stream_in(self.pg4, self.pg3)
- self.pg4.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=nat_ip_vrf10)
-
- # out2in VRF 10
- pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf10)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg4.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg4)
-
- # in2out VRF 20
- pkts = self.create_stream_in(self.pg6, self.pg3)
- self.pg6.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=nat_ip_vrf20)
-
- # out2in VRF 20
- pkts = self.create_stream_out(self.pg3, dst_ip=nat_ip_vrf20)
- self.pg3.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg6.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg6)
-
- def test_output_feature_hairpinning(self):
- """ S-NAT interface output feature hairpinning (in2out postrouting) """
- host = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
- host_in_port = 1234
- host_out_port = 0
- server_in_port = 5678
- server_out_port = 8765
-
- self.snat_add_address(self.snat_addr)
- self.vapi.snat_interface_add_del_output_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_output_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # add static mapping for server
- self.snat_add_static_mapping(server.ip4, self.snat_addr,
- server_in_port, server_out_port,
- proto=IP_PROTOS.tcp)
-
- # send packet from host to server
- p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
- IP(src=host.ip4, dst=self.snat_addr) /
- TCP(sport=host_in_port, dport=server_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
- self.assertEqual(ip.dst, server.ip4)
- self.assertNotEqual(tcp.sport, host_in_port)
- self.assertEqual(tcp.dport, server_in_port)
- self.check_tcp_checksum(p)
- host_out_port = tcp.sport
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- # send reply from server to host
- p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
- IP(src=server.ip4, dst=self.snat_addr) /
- TCP(sport=server_in_port, dport=host_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.snat_addr)
- self.assertEqual(ip.dst, host.ip4)
- self.assertEqual(tcp.sport, server_out_port)
- self.assertEqual(tcp.dport, host_in_port)
- self.check_tcp_checksum(p)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:"), p)
- raise
-
- def tearDown(self):
- super(TestSNAT, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show snat verbose"))
- self.clear_snat()
-
-
-class TestDeterministicNAT(MethodHolder):
- """ Deterministic NAT Test Cases """
-
- @classmethod
- def setUpConstants(cls):
- super(TestDeterministicNAT, cls).setUpConstants()
- cls.vpp_cmdline.extend(["snat", "{", "deterministic", "}"])
-
- @classmethod
- def setUpClass(cls):
- super(TestDeterministicNAT, cls).setUpClass()
-
- try:
- cls.tcp_port_in = 6303
- cls.tcp_external_port = 6303
- cls.udp_port_in = 6304
- cls.udp_external_port = 6304
- cls.icmp_id_in = 6305
- cls.snat_addr = '10.0.0.3'
-
- cls.create_pg_interfaces(range(3))
- cls.interfaces = list(cls.pg_interfaces)
-
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- cls.pg0.generate_remote_hosts(2)
- cls.pg0.configure_ipv4_neighbors()
-
- except Exception:
- super(TestDeterministicNAT, cls).tearDownClass()
- raise
-
- def create_stream_in(self, in_if, out_if, ttl=64):
- """
- Create packet stream for inside network
-
- :param in_if: Inside interface
- :param out_if: Outside interface
- :param ttl: TTL of generated packets
- """
- pkts = []
- # TCP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port))
- pkts.append(p)
-
- # UDP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- UDP(sport=self.udp_port_in, dport=self.udp_external_port))
- pkts.append(p)
-
- # ICMP
- p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
- ICMP(id=self.icmp_id_in, type='echo-request'))
- pkts.append(p)
-
- return pkts
-
- def create_stream_out(self, out_if, dst_ip=None, ttl=64):
- """
- Create packet stream for outside network
-
- :param out_if: Outside interface
- :param dst_ip: Destination IP address (Default use global SNAT address)
- :param ttl: TTL of generated packets
- """
- if dst_ip is None:
- dst_ip = self.snat_addr
- pkts = []
- # TCP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- TCP(dport=self.tcp_port_out, sport=self.tcp_external_port))
- pkts.append(p)
-
- # UDP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- UDP(dport=self.udp_port_out, sport=self.udp_external_port))
- pkts.append(p)
-
- # ICMP
- p = (Ether(dst=out_if.local_mac, src=out_if.remote_mac) /
- IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
- ICMP(id=self.icmp_external_id, type='echo-reply'))
- pkts.append(p)
-
- return pkts
-
- def verify_capture_out(self, capture, nat_ip=None, packet_num=3):
- """
- Verify captured packets on outside network
-
- :param capture: Captured packets
- :param nat_ip: Translated IP address (Default use global SNAT address)
- :param same_port: Sorce port number is not translated (Default False)
- :param packet_num: Expected number of packets (Default 3)
- """
- if nat_ip is None:
- nat_ip = self.snat_addr
- self.assertEqual(packet_num, len(capture))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, nat_ip)
- if packet.haslayer(TCP):
- self.tcp_port_out = packet[TCP].sport
- elif packet.haslayer(UDP):
- self.udp_port_out = packet[UDP].sport
- else:
- self.icmp_external_id = packet[ICMP].id
- except:
- self.logger.error(ppp("Unexpected or invalid packet "
- "(outside network):", packet))
- raise
-
- def initiate_tcp_session(self, in_if, out_if):
- """
- Initiates TCP session
-
- :param in_if: Inside interface
- :param out_if: Outside interface
- """
- try:
- # SYN packet in->out
- p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="S"))
- in_if.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = out_if.get_capture(1)
- p = capture[0]
- self.tcp_port_out = p[TCP].sport
-
- # SYN + ACK packet out->in
- p = (Ether(src=out_if.remote_mac, dst=out_if.local_mac) /
- IP(src=out_if.remote_ip4, dst=self.snat_addr) /
- TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
- flags="SA"))
- out_if.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- in_if.get_capture(1)
-
- # ACK packet in->out
- p = (Ether(src=in_if.remote_mac, dst=in_if.local_mac) /
- IP(src=in_if.remote_ip4, dst=out_if.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="A"))
- in_if.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- out_if.get_capture(1)
-
- except:
- self.logger.error("TCP 3 way handshake failed")
- raise
-
- def verify_ipfix_max_entries_per_user(self, data):
- """
- Verify IPFIX maximum entries per user exceeded event
-
- :param data: Decoded IPFIX data records
- """
- self.assertEqual(1, len(data))
- record = data[0]
- # natEvent
- self.assertEqual(ord(record[230]), 13)
- # natQuotaExceededEvent
- self.assertEqual('\x03\x00\x00\x00', record[466])
- # sourceIPv4Address
- self.assertEqual(self.pg0.remote_ip4n, record[8])
-
- def test_deterministic_mode(self):
- """ S-NAT run deterministic mode """
- in_addr = '172.16.255.0'
- out_addr = '172.17.255.50'
- in_addr_t = '172.16.255.20'
- in_addr_n = socket.inet_aton(in_addr)
- out_addr_n = socket.inet_aton(out_addr)
- in_addr_t_n = socket.inet_aton(in_addr_t)
- in_plen = 24
- out_plen = 32
-
- snat_config = self.vapi.snat_show_config()
- self.assertEqual(1, snat_config.deterministic)
-
- self.vapi.snat_add_det_map(in_addr_n, in_plen, out_addr_n, out_plen)
-
- rep1 = self.vapi.snat_det_forward(in_addr_t_n)
- self.assertEqual(rep1.out_addr[:4], out_addr_n)
- rep2 = self.vapi.snat_det_reverse(out_addr_n, rep1.out_port_hi)
- self.assertEqual(rep2.in_addr[:4], in_addr_t_n)
-
- deterministic_mappings = self.vapi.snat_det_map_dump()
- self.assertEqual(len(deterministic_mappings), 1)
- dsm = deterministic_mappings[0]
- self.assertEqual(in_addr_n, dsm.in_addr[:4])
- self.assertEqual(in_plen, dsm.in_plen)
- self.assertEqual(out_addr_n, dsm.out_addr[:4])
- self.assertEqual(out_plen, dsm.out_plen)
-
- self.clear_snat()
- deterministic_mappings = self.vapi.snat_det_map_dump()
- self.assertEqual(len(deterministic_mappings), 0)
-
- def test_set_timeouts(self):
- """ Set deterministic NAT timeouts """
- timeouts_before = self.vapi.snat_det_get_timeouts()
-
- self.vapi.snat_det_set_timeouts(timeouts_before.udp + 10,
- timeouts_before.tcp_established + 10,
- timeouts_before.tcp_transitory + 10,
- timeouts_before.icmp + 10)
-
- timeouts_after = self.vapi.snat_det_get_timeouts()
-
- self.assertNotEqual(timeouts_before.udp, timeouts_after.udp)
- self.assertNotEqual(timeouts_before.icmp, timeouts_after.icmp)
- self.assertNotEqual(timeouts_before.tcp_established,
- timeouts_after.tcp_established)
- self.assertNotEqual(timeouts_before.tcp_transitory,
- timeouts_after.tcp_transitory)
-
- def test_det_in(self):
- """ CGNAT translation test (TCP, UDP, ICMP) """
-
- nat_ip = "10.0.0.10"
-
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(nat_ip),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, nat_ip)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in(capture, self.pg0)
-
- # session dump test
- sessions = self.vapi.snat_det_session_dump(self.pg0.remote_ip4n)
- self.assertEqual(len(sessions), 3)
-
- # TCP session
- s = sessions[0]
- self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
- self.assertEqual(s.in_port, self.tcp_port_in)
- self.assertEqual(s.out_port, self.tcp_port_out)
- self.assertEqual(s.ext_port, self.tcp_external_port)
-
- # UDP session
- s = sessions[1]
- self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
- self.assertEqual(s.in_port, self.udp_port_in)
- self.assertEqual(s.out_port, self.udp_port_out)
- self.assertEqual(s.ext_port, self.udp_external_port)
-
- # ICMP session
- s = sessions[2]
- self.assertEqual(s.ext_addr[:4], self.pg1.remote_ip4n)
- self.assertEqual(s.in_port, self.icmp_id_in)
- self.assertEqual(s.out_port, self.icmp_external_id)
-
- def test_multiple_users(self):
- """ CGNAT multiple users """
-
- nat_ip = "10.0.0.10"
- port_in = 80
- external_port = 6303
-
- host0 = self.pg0.remote_hosts[0]
- host1 = self.pg0.remote_hosts[1]
-
- self.vapi.snat_add_det_map(host0.ip4n,
- 24,
- socket.inet_aton(nat_ip),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- # host0 to out
- p = (Ether(src=host0.mac, dst=self.pg0.local_mac) /
- IP(src=host0.ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=port_in, dport=external_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, nat_ip)
- self.assertEqual(ip.dst, self.pg1.remote_ip4)
- self.assertEqual(tcp.dport, external_port)
- port_out0 = tcp.sport
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- # host1 to out
- p = (Ether(src=host1.mac, dst=self.pg0.local_mac) /
- IP(src=host1.ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=port_in, dport=external_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, nat_ip)
- self.assertEqual(ip.dst, self.pg1.remote_ip4)
- self.assertEqual(tcp.dport, external_port)
- port_out1 = tcp.sport
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(1, len(dms))
- self.assertEqual(2, dms[0].ses_num)
-
- # out to host0
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=nat_ip) /
- TCP(sport=external_port, dport=port_out0))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.pg1.remote_ip4)
- self.assertEqual(ip.dst, host0.ip4)
- self.assertEqual(tcp.dport, port_in)
- self.assertEqual(tcp.sport, external_port)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", p))
- raise
-
- # out to host1
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=nat_ip) /
- TCP(sport=external_port, dport=port_out1))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(1)
- p = capture[0]
- try:
- ip = p[IP]
- tcp = p[TCP]
- self.assertEqual(ip.src, self.pg1.remote_ip4)
- self.assertEqual(ip.dst, host1.ip4)
- self.assertEqual(tcp.dport, port_in)
- self.assertEqual(tcp.sport, external_port)
- except:
- self.logger.error(ppp("Unexpected or invalid packet", p))
- raise
-
- # session close api test
- self.vapi.snat_det_close_session_out(socket.inet_aton(nat_ip),
- port_out1,
- self.pg1.remote_ip4n,
- external_port)
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(dms[0].ses_num, 1)
-
- self.vapi.snat_det_close_session_in(host0.ip4n,
- port_in,
- self.pg1.remote_ip4n,
- external_port)
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(dms[0].ses_num, 0)
-
- def test_tcp_session_close_detection_in(self):
- """ CGNAT TCP session close initiated from inside network """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- self.initiate_tcp_session(self.pg0, self.pg1)
-
- # close the session from inside
- try:
- # FIN packet in -> out
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="F"))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg1.get_capture(1)
-
- pkts = []
-
- # ACK packet out -> in
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
- flags="A"))
- pkts.append(p)
-
- # FIN packet out -> in
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
- flags="F"))
- pkts.append(p)
-
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg0.get_capture(2)
-
- # ACK packet in -> out
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="A"))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg1.get_capture(1)
-
- # Check if snat closed the session
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(0, dms[0].ses_num)
- except:
- self.logger.error("TCP session termination failed")
- raise
-
- def test_tcp_session_close_detection_out(self):
- """ CGNAT TCP session close initiated from outside network """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- self.initiate_tcp_session(self.pg0, self.pg1)
-
- # close the session from outside
- try:
- # FIN packet out -> in
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
- flags="F"))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg0.get_capture(1)
-
- pkts = []
-
- # ACK packet in -> out
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="A"))
- pkts.append(p)
-
- # ACK packet in -> out
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- TCP(sport=self.tcp_port_in, dport=self.tcp_external_port,
- flags="F"))
- pkts.append(p)
-
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg1.get_capture(2)
-
- # ACK packet out -> in
- p = (Ether(src=self.pg1.remote_mac, dst=self.pg1.local_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.snat_addr) /
- TCP(sport=self.tcp_external_port, dport=self.tcp_port_out,
- flags="A"))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg0.get_capture(1)
-
- # Check if snat closed the session
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(0, dms[0].ses_num)
- except:
- self.logger.error("TCP session termination failed")
- raise
-
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
- def test_session_timeout(self):
- """ CGNAT session timeouts """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
-
- self.initiate_tcp_session(self.pg0, self.pg1)
- self.vapi.snat_det_set_timeouts(5, 5, 5, 5)
- pkts = self.create_stream_in(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- sleep(15)
-
- dms = self.vapi.snat_det_map_dump()
- self.assertEqual(0, dms[0].ses_num)
-
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
- def test_session_limit_per_user(self):
- """ CGNAT maximum 1000 sessions per user should be created """
- self.vapi.snat_add_det_map(self.pg0.remote_ip4n,
- 32,
- socket.inet_aton(self.snat_addr),
- 32)
- self.vapi.snat_interface_add_del_feature(self.pg0.sw_if_index)
- self.vapi.snat_interface_add_del_feature(self.pg1.sw_if_index,
- is_inside=0)
- self.vapi.set_ipfix_exporter(collector_address=self.pg2.remote_ip4n,
- src_address=self.pg2.local_ip4n,
- path_mtu=512,
- template_interval=10)
- self.vapi.snat_ipfix()
-
- pkts = []
- for port in range(1025, 2025):
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(sport=port, dport=port))
- pkts.append(p)
-
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
-
- p = (Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac) /
- IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4) /
- UDP(sport=3001, dport=3002))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.assert_nothing_captured()
-
- # verify ICMP error packet
- capture = self.pg0.get_capture(1)
- p = capture[0]
- self.assertTrue(p.haslayer(ICMP))
- icmp = p[ICMP]
- self.assertEqual(icmp.type, 3)
- self.assertEqual(icmp.code, 1)
- self.assertTrue(icmp.haslayer(IPerror))
- inner_ip = icmp[IPerror]
- self.assertEqual(inner_ip[UDPerror].sport, 3001)
- self.assertEqual(inner_ip[UDPerror].dport, 3002)
-
- dms = self.vapi.snat_det_map_dump()
-
- self.assertEqual(1000, dms[0].ses_num)
-
- # verify IPFIX logging
- self.vapi.cli("ipfix flush") # FIXME this should be an API call
- sleep(1)
- capture = self.pg2.get_capture(2)
- ipfix = IPFIXDecoder()
- # first load template
- for p in capture:
- self.assertTrue(p.haslayer(IPFIX))
- if p.haslayer(Template):
- ipfix.add_template(p.getlayer(Template))
- # verify events in data set
- for p in capture:
- if p.haslayer(Data):
- data = ipfix.decode_data_set(p.getlayer(Set))
- self.verify_ipfix_max_entries_per_user(data)
-
- def clear_snat(self):
- """
- Clear SNAT configuration.
- """
- self.vapi.snat_ipfix(enable=0)
- self.vapi.snat_det_set_timeouts()
- deterministic_mappings = self.vapi.snat_det_map_dump()
- for dsm in deterministic_mappings:
- self.vapi.snat_add_det_map(dsm.in_addr,
- dsm.in_plen,
- dsm.out_addr,
- dsm.out_plen,
- is_add=0)
-
- interfaces = self.vapi.snat_interface_dump()
- for intf in interfaces:
- self.vapi.snat_interface_add_del_feature(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
-
- def tearDown(self):
- super(TestDeterministicNAT, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show snat detail"))
- self.clear_snat()
-
-
-class TestNAT64(MethodHolder):
- """ NAT64 Test Cases """
-
- @classmethod
- def setUpClass(cls):
- super(TestNAT64, cls).setUpClass()
-
- try:
- cls.tcp_port_in = 6303
- cls.tcp_port_out = 6303
- cls.udp_port_in = 6304
- cls.udp_port_out = 6304
- cls.icmp_id_in = 6305
- cls.icmp_id_out = 6305
- cls.nat_addr = '10.0.0.3'
- cls.nat_addr_n = socket.inet_pton(socket.AF_INET, cls.nat_addr)
- cls.vrf1_id = 10
- cls.vrf1_nat_addr = '10.0.10.3'
- cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
- cls.vrf1_nat_addr)
-
- cls.create_pg_interfaces(range(3))
- cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
- cls.ip6_interfaces.append(cls.pg_interfaces[2])
- cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
-
- cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
-
- cls.pg0.generate_remote_hosts(2)
-
- for i in cls.ip6_interfaces:
- i.admin_up()
- i.config_ip6()
- i.configure_ipv6_neighbors()
-
- for i in cls.ip4_interfaces:
- i.admin_up()
- i.config_ip4()
- i.resolve_arp()
-
- except Exception:
- super(TestNAT64, cls).tearDownClass()
- raise
-
- def test_pool(self):
- """ Add/delete address to NAT64 pool """
- nat_addr = socket.inet_pton(socket.AF_INET, '1.2.3.4')
-
- self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr)
-
- addresses = self.vapi.nat64_pool_addr_dump()
- self.assertEqual(len(addresses), 1)
- self.assertEqual(addresses[0].address, nat_addr)
-
- self.vapi.nat64_add_del_pool_addr_range(nat_addr, nat_addr, is_add=0)
-
- addresses = self.vapi.nat64_pool_addr_dump()
- self.assertEqual(len(addresses), 0)
-
- def test_interface(self):
- """ Enable/disable NAT64 feature on the interface """
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- interfaces = self.vapi.nat64_interface_dump()
- self.assertEqual(len(interfaces), 2)
- pg0_found = False
- pg1_found = False
- for intf in interfaces:
- if intf.sw_if_index == self.pg0.sw_if_index:
- self.assertEqual(intf.is_inside, 1)
- pg0_found = True
- elif intf.sw_if_index == self.pg1.sw_if_index:
- self.assertEqual(intf.is_inside, 0)
- pg1_found = True
- self.assertTrue(pg0_found)
- self.assertTrue(pg1_found)
-
- features = self.vapi.cli("show interface features pg0")
- self.assertNotEqual(features.find('nat64-in2out'), -1)
- features = self.vapi.cli("show interface features pg1")
- self.assertNotEqual(features.find('nat64-out2in'), -1)
-
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index, is_add=0)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_add=0)
-
- interfaces = self.vapi.nat64_interface_dump()
- self.assertEqual(len(interfaces), 0)
-
- def test_static_bib(self):
- """ Add/delete static BIB entry """
- in_addr = socket.inet_pton(socket.AF_INET6,
- '2001:db8:85a3::8a2e:370:7334')
- out_addr = socket.inet_pton(socket.AF_INET, '10.1.1.3')
- in_port = 1234
- out_port = 5678
- proto = IP_PROTOS.tcp
-
- self.vapi.nat64_add_del_static_bib(in_addr,
- out_addr,
- in_port,
- out_port,
- proto)
- bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
- static_bib_num = 0
- for bibe in bib:
- if bibe.is_static:
- static_bib_num += 1
- self.assertEqual(bibe.i_addr, in_addr)
- self.assertEqual(bibe.o_addr, out_addr)
- self.assertEqual(bibe.i_port, in_port)
- self.assertEqual(bibe.o_port, out_port)
- self.assertEqual(static_bib_num, 1)
-
- self.vapi.nat64_add_del_static_bib(in_addr,
- out_addr,
- in_port,
- out_port,
- proto,
- is_add=0)
- bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
- static_bib_num = 0
- for bibe in bib:
- if bibe.is_static:
- static_bib_num += 1
- self.assertEqual(static_bib_num, 0)
-
- def test_set_timeouts(self):
- """ Set NAT64 timeouts """
- # verify default values
- timeouts = self.vapi.nat64_get_timeouts()
- self.assertEqual(timeouts.udp, 300)
- self.assertEqual(timeouts.icmp, 60)
- self.assertEqual(timeouts.tcp_trans, 240)
- self.assertEqual(timeouts.tcp_est, 7440)
- self.assertEqual(timeouts.tcp_incoming_syn, 6)
-
- # set and verify custom values
- self.vapi.nat64_set_timeouts(udp=200, icmp=30, tcp_trans=250,
- tcp_est=7450, tcp_incoming_syn=10)
- timeouts = self.vapi.nat64_get_timeouts()
- self.assertEqual(timeouts.udp, 200)
- self.assertEqual(timeouts.icmp, 30)
- self.assertEqual(timeouts.tcp_trans, 250)
- self.assertEqual(timeouts.tcp_est, 7450)
- self.assertEqual(timeouts.tcp_incoming_syn, 10)
-
- def test_dynamic(self):
- """ NAT64 dynamic translation test """
- self.tcp_port_in = 6303
- self.udp_port_in = 6304
- self.icmp_id_in = 6305
-
- ses_num_start = self.nat64_get_ses_num()
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- # in2out
- pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
- self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
-
- # in2out
- pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
-
- ses_num_end = self.nat64_get_ses_num()
-
- self.assertEqual(ses_num_end - ses_num_start, 3)
-
- # tenant with specific VRF
- self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
- self.vrf1_nat_addr_n,
- vrf_id=self.vrf1_id)
- self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
-
- pkts = self.create_stream_in_ip6(self.pg2, self.pg1)
- self.pg2.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg2.remote_ip6)
-
- def test_static(self):
- """ NAT64 static translation test """
- self.tcp_port_in = 60303
- self.udp_port_in = 60304
- self.icmp_id_in = 60305
- self.tcp_port_out = 60303
- self.udp_port_out = 60304
- self.icmp_id_out = 60305
-
- ses_num_start = self.nat64_get_ses_num()
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
- self.nat_addr_n,
- self.tcp_port_in,
- self.tcp_port_out,
- IP_PROTOS.tcp)
- self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
- self.nat_addr_n,
- self.udp_port_in,
- self.udp_port_out,
- IP_PROTOS.udp)
- self.vapi.nat64_add_del_static_bib(self.pg0.remote_ip6n,
- self.nat_addr_n,
- self.icmp_id_in,
- self.icmp_id_out,
- IP_PROTOS.icmp)
-
- # in2out
- pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.nat_addr,
- dst_ip=self.pg1.remote_ip4, same_port=True)
-
- # out2in
- pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
- self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
-
- ses_num_end = self.nat64_get_ses_num()
-
- self.assertEqual(ses_num_end - ses_num_start, 3)
-
- @unittest.skipUnless(running_extended_tests(), "part of extended tests")
- def test_session_timeout(self):
- """ NAT64 session timeout """
- self.icmp_id_in = 1234
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
- self.vapi.nat64_set_timeouts(icmp=5)
-
- pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
-
- ses_num_before_timeout = self.nat64_get_ses_num()
-
- sleep(15)
-
- # ICMP session after timeout
- ses_num_after_timeout = self.nat64_get_ses_num()
- self.assertNotEqual(ses_num_before_timeout, ses_num_after_timeout)
-
- def test_icmp_error(self):
- """ NAT64 ICMP Error message translation """
- self.tcp_port_in = 6303
- self.udp_port_in = 6304
- self.icmp_id_in = 6305
-
- ses_num_start = self.nat64_get_ses_num()
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- # send some packets to create sessions
- pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture_ip4 = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture_ip4,
- nat_ip=self.nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture_ip6 = self.pg0.get_capture(len(pkts))
- ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
- self.verify_capture_in_ip6(capture_ip6, ip[IPv6].src,
- self.pg0.remote_ip6)
-
- # in2out
- pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=ip[IPv6].src) /
- ICMPv6DestUnreach(code=1) /
- packet[IPv6] for packet in capture_ip6]
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IP].src, self.nat_addr)
- self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertEqual(packet[ICMP].type, 3)
- self.assertEqual(packet[ICMP].code, 13)
- inner = packet[IPerror]
- self.assertEqual(inner.src, self.pg1.remote_ip4)
- self.assertEqual(inner.dst, self.nat_addr)
- self.check_icmp_checksum(packet)
- if inner.haslayer(TCPerror):
- self.assertEqual(inner[TCPerror].dport, self.tcp_port_out)
- elif inner.haslayer(UDPerror):
- self.assertEqual(inner[UDPerror].dport, self.udp_port_out)
- else:
- self.assertEqual(inner[ICMPerror].id, self.icmp_id_out)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # out2in
- pkts = [Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
- ICMP(type=3, code=13) /
- packet[IP] for packet in capture_ip4]
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, ip.src)
- self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
- icmp = packet[ICMPv6DestUnreach]
- self.assertEqual(icmp.code, 1)
- inner = icmp[IPerror6]
- self.assertEqual(inner.src, self.pg0.remote_ip6)
- self.assertEqual(inner.dst, ip.src)
- self.check_icmpv6_checksum(packet)
- if inner.haslayer(TCPerror):
- self.assertEqual(inner[TCPerror].sport, self.tcp_port_in)
- elif inner.haslayer(UDPerror):
- self.assertEqual(inner[UDPerror].sport, self.udp_port_in)
- else:
- self.assertEqual(inner[ICMPv6EchoRequest].id,
- self.icmp_id_in)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_hairpinning(self):
- """ NAT64 hairpinning """
-
- client = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
- server_tcp_in_port = 22
- server_tcp_out_port = 4022
- server_udp_in_port = 23
- server_udp_out_port = 4023
- client_tcp_in_port = 1234
- client_udp_in_port = 1235
- client_tcp_out_port = 0
- client_udp_out_port = 0
- ip = IPv6(src=''.join(['64:ff9b::', self.nat_addr]))
- nat_addr_ip6 = ip.src
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- self.vapi.nat64_add_del_static_bib(server.ip6n,
- self.nat_addr_n,
- server_tcp_in_port,
- server_tcp_out_port,
- IP_PROTOS.tcp)
- self.vapi.nat64_add_del_static_bib(server.ip6n,
- self.nat_addr_n,
- server_udp_in_port,
- server_udp_out_port,
- IP_PROTOS.udp)
-
- # client to server
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=client.ip6, dst=nat_addr_ip6) /
- TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=client.ip6, dst=nat_addr_ip6) /
- UDP(sport=client_udp_in_port, dport=server_udp_out_port))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, nat_addr_ip6)
- self.assertEqual(packet[IPv6].dst, server.ip6)
- if packet.haslayer(TCP):
- self.assertNotEqual(packet[TCP].sport, client_tcp_in_port)
- self.assertEqual(packet[TCP].dport, server_tcp_in_port)
- self.check_tcp_checksum(packet)
- client_tcp_out_port = packet[TCP].sport
- else:
- self.assertNotEqual(packet[UDP].sport, client_udp_in_port)
- self.assertEqual(packet[UDP].dport, server_udp_in_port)
- self.check_udp_checksum(packet)
- client_udp_out_port = packet[UDP].sport
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server to client
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=server.ip6, dst=nat_addr_ip6) /
- TCP(sport=server_tcp_in_port, dport=client_tcp_out_port))
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=server.ip6, dst=nat_addr_ip6) /
- UDP(sport=server_udp_in_port, dport=client_udp_out_port))
- pkts.append(p)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, nat_addr_ip6)
- self.assertEqual(packet[IPv6].dst, client.ip6)
- if packet.haslayer(TCP):
- self.assertEqual(packet[TCP].sport, server_tcp_out_port)
- self.assertEqual(packet[TCP].dport, client_tcp_in_port)
- self.check_tcp_checksum(packet)
- else:
- self.assertEqual(packet[UDP].sport, server_udp_out_port)
- self.assertEqual(packet[UDP].dport, client_udp_in_port)
- self.check_udp_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # ICMP error
- pkts = []
- pkts = [Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=client.ip6, dst=nat_addr_ip6) /
- ICMPv6DestUnreach(code=1) /
- packet[IPv6] for packet in capture]
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, nat_addr_ip6)
- self.assertEqual(packet[IPv6].dst, server.ip6)
- icmp = packet[ICMPv6DestUnreach]
- self.assertEqual(icmp.code, 1)
- inner = icmp[IPerror6]
- self.assertEqual(inner.src, server.ip6)
- self.assertEqual(inner.dst, nat_addr_ip6)
- self.check_icmpv6_checksum(packet)
- if inner.haslayer(TCPerror):
- self.assertEqual(inner[TCPerror].sport, server_tcp_in_port)
- self.assertEqual(inner[TCPerror].dport,
- client_tcp_out_port)
- else:
- self.assertEqual(inner[UDPerror].sport, server_udp_in_port)
- self.assertEqual(inner[UDPerror].dport,
- client_udp_out_port)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_prefix(self):
- """ NAT64 Network-Specific Prefix """
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
- self.vapi.nat64_add_del_pool_addr_range(self.vrf1_nat_addr_n,
- self.vrf1_nat_addr_n,
- vrf_id=self.vrf1_id)
- self.vapi.nat64_add_del_interface(self.pg2.sw_if_index)
-
- # Add global prefix
- global_pref64 = "2001:db8::"
- global_pref64_n = socket.inet_pton(socket.AF_INET6, global_pref64)
- global_pref64_len = 32
- self.vapi.nat64_add_del_prefix(global_pref64_n, global_pref64_len)
-
- prefix = self.vapi.nat64_prefix_dump()
- self.assertEqual(len(prefix), 1)
- self.assertEqual(prefix[0].prefix, global_pref64_n)
- self.assertEqual(prefix[0].prefix_len, global_pref64_len)
- self.assertEqual(prefix[0].vrf_id, 0)
-
- # Add tenant specific prefix
- vrf1_pref64 = "2001:db8:122:300::"
- vrf1_pref64_n = socket.inet_pton(socket.AF_INET6, vrf1_pref64)
- vrf1_pref64_len = 56
- self.vapi.nat64_add_del_prefix(vrf1_pref64_n,
- vrf1_pref64_len,
- vrf_id=self.vrf1_id)
- prefix = self.vapi.nat64_prefix_dump()
- self.assertEqual(len(prefix), 2)
-
- # Global prefix
- pkts = self.create_stream_in_ip6(self.pg0,
- self.pg1,
- pref=global_pref64,
- plen=global_pref64_len)
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg0.get_capture(len(pkts))
- dst_ip = self.compose_ip6(self.pg1.remote_ip4,
- global_pref64,
- global_pref64_len)
- self.verify_capture_in_ip6(capture, dst_ip, self.pg0.remote_ip6)
-
- # Tenant specific prefix
- pkts = self.create_stream_in_ip6(self.pg2,
- self.pg1,
- pref=vrf1_pref64,
- plen=vrf1_pref64_len)
- self.pg2.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture, nat_ip=self.vrf1_nat_addr,
- dst_ip=self.pg1.remote_ip4)
-
- pkts = self.create_stream_out(self.pg1, dst_ip=self.vrf1_nat_addr)
- self.pg1.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg2.get_capture(len(pkts))
- dst_ip = self.compose_ip6(self.pg1.remote_ip4,
- vrf1_pref64,
- vrf1_pref64_len)
- self.verify_capture_in_ip6(capture, dst_ip, self.pg2.remote_ip6)
-
- def test_unknown_proto(self):
- """ NAT64 translate packet with unknown protocol """
-
- self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
- self.nat_addr_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
- remote_ip6 = self.compose_ip6(self.pg1.remote_ip4, '64:ff9b::', 96)
-
- # in2out
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=remote_ip6) /
- TCP(sport=self.tcp_port_in, dport=20))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg1.get_capture(1)
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=remote_ip6, nh=47) /
- GRE() /
- IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg1.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IP].src, self.nat_addr)
- self.assertEqual(packet[IP].dst, self.pg1.remote_ip4)
- self.assertTrue(packet.haslayer(GRE))
- self.check_ip_checksum(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # out2in
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IP(src=self.pg1.remote_ip4, dst=self.nat_addr) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg1.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IPv6].src, remote_ip6)
- self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
- self.assertEqual(packet[IPv6].nh, 47)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def test_hairpinning_unknown_proto(self):
- """ NAT64 translate packet with unknown protocol - hairpinning """
-
- client = self.pg0.remote_hosts[0]
- server = self.pg0.remote_hosts[1]
- server_tcp_in_port = 22
- server_tcp_out_port = 4022
- client_tcp_in_port = 1234
- client_tcp_out_port = 1235
- server_nat_ip = "10.0.0.100"
- client_nat_ip = "10.0.0.110"
- server_nat_ip_n = socket.inet_pton(socket.AF_INET, server_nat_ip)
- client_nat_ip_n = socket.inet_pton(socket.AF_INET, client_nat_ip)
- server_nat_ip6 = self.compose_ip6(server_nat_ip, '64:ff9b::', 96)
- client_nat_ip6 = self.compose_ip6(client_nat_ip, '64:ff9b::', 96)
-
- self.vapi.nat64_add_del_pool_addr_range(server_nat_ip_n,
- client_nat_ip_n)
- self.vapi.nat64_add_del_interface(self.pg0.sw_if_index)
- self.vapi.nat64_add_del_interface(self.pg1.sw_if_index, is_inside=0)
-
- self.vapi.nat64_add_del_static_bib(server.ip6n,
- server_nat_ip_n,
- server_tcp_in_port,
- server_tcp_out_port,
- IP_PROTOS.tcp)
-
- self.vapi.nat64_add_del_static_bib(server.ip6n,
- server_nat_ip_n,
- 0,
- 0,
- IP_PROTOS.gre)
-
- self.vapi.nat64_add_del_static_bib(client.ip6n,
- client_nat_ip_n,
- client_tcp_in_port,
- client_tcp_out_port,
- IP_PROTOS.tcp)
-
- # client to server
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=client.ip6, dst=server_nat_ip6) /
- TCP(sport=client_tcp_in_port, dport=server_tcp_out_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
-
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=client.ip6, dst=server_nat_ip6, nh=IP_PROTOS.gre) /
- GRE() /
- IP(src=self.pg2.local_ip4, dst=self.pg2.remote_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IPv6].src, client_nat_ip6)
- self.assertEqual(packet[IPv6].dst, server.ip6)
- self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # server to client
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=server.ip6, dst=client_nat_ip6, nh=IP_PROTOS.gre) /
- GRE() /
- IP(src=self.pg2.remote_ip4, dst=self.pg2.local_ip4) /
- TCP(sport=1234, dport=1234))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- p = self.pg0.get_capture(1)
- packet = p[0]
- try:
- self.assertEqual(packet[IPv6].src, server_nat_ip6)
- self.assertEqual(packet[IPv6].dst, client.ip6)
- self.assertEqual(packet[IPv6].nh, IP_PROTOS.gre)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def nat64_get_ses_num(self):
- """
- Return number of active NAT64 sessions.
- """
- st = self.vapi.nat64_st_dump()
- return len(st)
-
- def clear_nat64(self):
- """
- Clear NAT64 configuration.
- """
- self.vapi.nat64_set_timeouts()
-
- interfaces = self.vapi.nat64_interface_dump()
- for intf in interfaces:
- self.vapi.nat64_add_del_interface(intf.sw_if_index,
- intf.is_inside,
- is_add=0)
-
- bib = self.vapi.nat64_bib_dump(IP_PROTOS.tcp)
- for bibe in bib:
- if bibe.is_static:
- self.vapi.nat64_add_del_static_bib(bibe.i_addr,
- bibe.o_addr,
- bibe.i_port,
- bibe.o_port,
- bibe.proto,
- bibe.vrf_id,
- is_add=0)
-
- bib = self.vapi.nat64_bib_dump(IP_PROTOS.udp)
- for bibe in bib:
- if bibe.is_static:
- self.vapi.nat64_add_del_static_bib(bibe.i_addr,
- bibe.o_addr,
- bibe.i_port,
- bibe.o_port,
- bibe.proto,
- bibe.vrf_id,
- is_add=0)
-
- bib = self.vapi.nat64_bib_dump(IP_PROTOS.icmp)
- for bibe in bib:
- if bibe.is_static:
- self.vapi.nat64_add_del_static_bib(bibe.i_addr,
- bibe.o_addr,
- bibe.i_port,
- bibe.o_port,
- bibe.proto,
- bibe.vrf_id,
- is_add=0)
-
- adresses = self.vapi.nat64_pool_addr_dump()
- for addr in adresses:
- self.vapi.nat64_add_del_pool_addr_range(addr.address,
- addr.address,
- vrf_id=addr.vrf_id,
- is_add=0)
-
- prefixes = self.vapi.nat64_prefix_dump()
- for prefix in prefixes:
- self.vapi.nat64_add_del_prefix(prefix.prefix,
- prefix.prefix_len,
- vrf_id=prefix.vrf_id,
- is_add=0)
-
- def tearDown(self):
- super(TestNAT64, self).tearDown()
- if not self.vpp_dead:
- self.logger.info(self.vapi.cli("show nat64 pool"))
- self.logger.info(self.vapi.cli("show nat64 interfaces"))
- self.logger.info(self.vapi.cli("show nat64 prefix"))
- self.logger.info(self.vapi.cli("show nat64 bib all"))
- self.logger.info(self.vapi.cli("show nat64 session table all"))
- self.clear_nat64()
-
-if __name__ == '__main__':
- unittest.main(testRunner=VppTestRunner)