import socket
import unittest
-from logging import *
-
-from framework import VppTestCase, VppTestRunner
+import struct
+from framework import VppTestCase, VppTestRunner, running_extended_tests
from scapy.layers.inet import IP, TCP, UDP, ICMP
-from scapy.layers.l2 import Ether
+from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
+from scapy.layers.l2 import Ether, ARP
+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 TestSNAT(VppTestCase):
- """ SNAT Test Cases """
+class MethodHolder(VppTestCase):
+ """ SNAT create capture and verify method holder """
@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.create_pg_interfaces(range(7))
- 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(2)
- cls.pg0.configure_ipv4_neighbors()
-
- cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
+ super(MethodHolder, cls).setUpClass()
- for i in cls.overlapping_interfaces:
- i._local_ip4 = "172.16.255.1"
- i._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
- i._remote_hosts[0]._ip4 = "172.16.255.2"
- i.set_table_ip4(i.sw_if_index)
- i.config_ip4()
- i.admin_up()
- i.resolve_arp()
-
- except Exception:
- super(TestSNAT, cls).tearDownClass()
- raise
+ def tearDown(self):
+ super(MethodHolder, self).tearDown()
- def create_stream_in(self, in_if, out_if):
+ 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) /
+ IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
TCP(sport=self.tcp_port_in))
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) /
+ IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) /
UDP(sport=self.udp_port_in))
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) /
+ 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):
+ 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
+ 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) /
+ IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
TCP(dport=self.tcp_port_out))
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) /
+ IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
UDP(dport=self.udp_port_out))
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) /
+ IP(src=out_if.remote_ip4, dst=dst_ip, ttl=ttl) /
ICMP(id=self.icmp_id_out, type='echo-reply'))
pkts.append(p)
if same_port:
self.assertEqual(packet[TCP].sport, self.tcp_port_in)
else:
- self.assertNotEqual(packet[TCP].sport, self.tcp_port_in)
+ self.assertNotEqual(
+ packet[TCP].sport, self.tcp_port_in)
self.tcp_port_out = packet[TCP].sport
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.assertNotEqual(
+ packet[UDP].sport, self.udp_port_in)
self.udp_port_out = packet[UDP].sport
else:
if same_port:
self.assertNotEqual(packet[ICMP].id, self.icmp_id_in)
self.icmp_id_out = packet[ICMP].id
except:
- error("Unexpected or invalid packet (outside network):")
- error(packet.show())
+ self.logger.error(ppp("Unexpected or invalid packet "
+ "(outside network):", packet))
raise
def verify_capture_in(self, capture, in_if, packet_num=3):
else:
self.assertEqual(packet[ICMP].id, self.icmp_id_in)
except:
- error("Unexpected or invalid packet (inside network):")
- error(packet.show())
+ 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(2)
+ 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.16.255.3"
+ cls.pg5._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
+ cls.pg5._remote_hosts[0]._ip4 = "172.16.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,
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()
addr.ip_address,
is_add=0)
- def snat_add_static_mapping(self, local_ip, external_ip, local_port=0,
- external_port=0, vrf_id=0, is_add=1):
+ 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 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, local_port, external_port,
- addr_only, vrf_id, is_add)
-
- def snat_add_address(self, ip, is_add=1):
+ 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 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)
+ 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.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ 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 """
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture, nat_ip, True)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
def test_static_out(self):
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
# in2out
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture, nat_ip, True)
def test_static_with_port_in(self):
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)
+ 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)
+ 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)
+ 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)
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
def test_static_with_port_out(self):
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)
+ 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)
+ 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)
+ 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)
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
# in2out
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
def test_static_vrf_aware(self):
self.icmp_id_out = 6305
self.snat_add_static_mapping(self.pg4.remote_ip4, nat_ip1,
- vrf_id=self.pg4.sw_if_index)
+ vrf_id=10)
self.snat_add_static_mapping(self.pg0.remote_ip4, nat_ip2,
- vrf_id=self.pg4.sw_if_index)
+ 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.pg4.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg3.get_capture()
+ 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
self.pg0.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg3.get_capture()
- self.verify_capture_out(capture, packet_num=0)
+ self.pg3.assert_nothing_captured()
def test_multiple_inside_interfaces(self):
- """ SNAT multiple inside interfaces with non-overlapping address space """
+ """ 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.pg2.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()
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in 1st interface
self.pg3.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
+ capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
# in2out 2nd interface
self.pg1.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg3.get_capture()
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in 2nd interface
self.pg3.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg1.get_capture()
+ capture = self.pg1.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg1)
- # in2out 3rd interface
- pkts = self.create_stream_in(self.pg2, self.pg3)
- self.pg2.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg3.get_capture()
- self.verify_capture_out(capture)
-
- # out2in 3rd 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.pg2.get_capture()
- self.verify_capture_in(capture, self.pg2)
-
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()
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in 1st interface
self.pg3.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg4.get_capture()
+ capture = self.pg4.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg4)
# in2out 2nd interface
self.pg5.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg3.get_capture()
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in 2nd interface
self.pg3.add_stream(pkts)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg5.get_capture()
+ 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()
- self.verify_capture_out(capture)
+ 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)
+ 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()
+ 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 """
is_inside=0)
# add static mapping for server
self.snat_add_static_mapping(server.ip4, self.snat_addr,
- server_in_port, server_out_port)
+ 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) /
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
- self.assertEqual(1, len(capture))
+ capture = self.pg0.get_capture(1)
p = capture[0]
try:
ip = p[IP]
self.assertEqual(tcp.dport, server_in_port)
host_out_port = tcp.sport
except:
- error("Unexpected or invalid packet:")
- error(p.show())
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
# send reply from server to host
self.pg0.add_stream(p)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
- capture = self.pg0.get_capture()
- self.assertEqual(1, len(capture))
+ capture = self.pg0.get_capture(1)
p = capture[0]
try:
ip = p[IP]
self.assertEqual(tcp.sport, server_out_port)
self.assertEqual(tcp.dport, host_in_port)
except:
- error("Unexpected or invalid packet:")
- error(p.show())
+ self.logger.error(ppp("Unexpected or invalid packet:"), p)
raise
+ def test_max_translations_per_user(self):
+ """ MAX translations per user - recycle the least recently used """
- def tearDown(self):
- super(TestSNAT, self).tearDown()
+ 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 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)
+
+ 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
+ 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()
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff