Gitiles
Code Review Sign In
gerrit.nordix.org / fdio / vpp / acd4c63e3c6e70ea3f58527d9bace7c0e38df719 / . / test / test_snat.py
blob: c2f9280d62aea418d3b91a3a3b0b77c247102e98 [file] [log] [blame]
#!/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
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_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 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 create_stream_in_ip6(self, in_if, out_if, hlim=64):
"""
Create IPv6 packet stream for inside network
:param in_if: Inside interface
:param out_if: Outside interface
:param ttl: Hop Limit of generated packets
"""
pkts = []
dst = ''.join(['64:ff9b::', out_if.remote_ip4])
# 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.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
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
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.assertEqual(packet[IP].dst, in_if.remote_ip4)
if packet.haslayer(TCP):
self.assertEqual(packet[TCP].dport, self.tcp_port_in)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, 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_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)
elif packet.haslayer(UDP):
self.assertEqual(packet[UDP].dport, self.udp_port_in)
else:
self.assertEqual(packet[ICMPv6EchoReply].id,
self.icmp_id_in)
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.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,
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 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()
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.create_pg_interfaces(range(2))
cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
for i in cls.ip6_interfaces:
i.admin_up()
i.config_ip6()
i.resolve_ndp()
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(3)
self.verify_capture_out(capture, packet_num=3, 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(3)
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(3)
self.verify_capture_out(capture, packet_num=3, 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(3)
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)
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(3)
self.verify_capture_out(capture, packet_num=3, 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(3)
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(3)
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, packet_num=3,
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)
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)
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 nat64_get_ses_num(self):
"""
Return number of active NAT64 sessions.
"""
ses_num = 0
st = self.vapi.nat64_st_dump(IP_PROTOS.tcp)
ses_num += len(st)
st = self.vapi.nat64_st_dump(IP_PROTOS.udp)
ses_num += len(st)
st = self.vapi.nat64_st_dump(IP_PROTOS.icmp)
ses_num += len(st)
return ses_num
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,
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 bib tcp"))
self.logger.info(self.vapi.cli("show nat64 bib udp"))
self.logger.info(self.vapi.cli("show nat64 bib icmp"))
self.logger.info(self.vapi.cli("show nat64 session table tcp"))
self.logger.info(self.vapi.cli("show nat64 session table udp"))
self.logger.info(self.vapi.cli("show nat64 session table icmp"))
self.clear_nat64()
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)