#!/usr/bin/env python3
+import ipaddress
+import random
import socket
-import unittest
import struct
-import random
-
-from framework import VppTestCase, VppTestRunner, running_extended_tests
+import unittest
+from io import BytesIO
+from time import sleep
import scapy.compat
+from framework import VppTestCase, VppTestRunner, running_extended_tests
+from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
+from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
+ IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
+ PacketListField
+from scapy.data import IP_PROTOS
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
+from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, \
ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6
-from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
from scapy.layers.l2 import Ether, ARP, GRE
-from scapy.data import IP_PROTOS
-from scapy.packet import bind_layers, Raw
-from util import ppp
-from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
-from time import sleep
-from util import ip4_range
-from vpp_papi import mac_pton
+from scapy.packet import Raw
from syslog_rfc5424_parser import SyslogMessage, ParseError
-from syslog_rfc5424_parser.constants import SyslogFacility, SyslogSeverity
-from io import BytesIO
-from vpp_papi import VppEnum
-from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
-from vpp_neighbor import VppNeighbor
-from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
- IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
- PacketListField
-from ipaddress import IPv6Network
+from syslog_rfc5424_parser.constants import SyslogSeverity
+from util import ip4_range
from util import ppc, ppp
-from socket import inet_pton, AF_INET
from vpp_acl import AclRule, VppAcl, VppAclInterface
+from vpp_ip_route import VppIpRoute, VppRoutePath
+from vpp_neighbor import VppNeighbor
+from vpp_papi import VppEnum
# NAT HA protocol event data
fields_desc = [ByteEnumField("event_type", None,
{1: "add", 2: "del", 3: "refresh"}),
ByteEnumField("protocol", None,
- {0: "udp", 1: "tcp", 2: "icmp"}),
+ {0: "other", 1: "udp", 2: "tcp", 3: "icmp"}),
ShortField("flags", 0),
IPField("in_addr", None),
IPField("out_addr", None),
return pkts
def verify_capture_out(self, capture, nat_ip=None, same_port=False,
- dst_ip=None, is_ip6=False):
+ dst_ip=None, is_ip6=False, ignore_port=False):
"""
Verify captured packets on outside network
if dst_ip is not None:
self.assertEqual(packet[IP46].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)
+ if not ignore_port:
+ if same_port:
+ self.assertEqual(
+ packet[TCP].sport, self.tcp_port_in)
+ else:
+ self.assertNotEqual(
+ packet[TCP].sport, self.tcp_port_in)
self.tcp_port_out = packet[TCP].sport
self.assert_packet_checksums_valid(packet)
elif packet.haslayer(UDP):
- if same_port:
- self.assertEqual(packet[UDP].sport, self.udp_port_in)
- else:
- self.assertNotEqual(
- packet[UDP].sport, self.udp_port_in)
+ if not ignore_port:
+ 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[ICMP46].id, self.icmp_id_in)
- else:
- self.assertNotEqual(packet[ICMP46].id, self.icmp_id_in)
+ if not ignore_port:
+ if same_port:
+ self.assertEqual(
+ packet[ICMP46].id, self.icmp_id_in)
+ else:
+ self.assertNotEqual(
+ packet[ICMP46].id, self.icmp_id_in)
self.icmp_id_out = packet[ICMP46].id
self.assert_packet_checksums_valid(packet)
except:
else:
nat44_ses_delete_num += 1
# sourceIPv4Address
- self.assertEqual(self.pg0.remote_ip4n, record[8])
+ self.assertEqual(self.pg0.remote_ip4,
+ str(ipaddress.IPv4Address(record[8])))
# postNATSourceIPv4Address
self.assertEqual(socket.inet_pton(socket.AF_INET, self.nat_addr),
record[225])
# maxBIBEntries
self.assertEqual(struct.pack("I", limit), record[472])
- def verify_ipfix_max_fragments_ip6(self, data, limit, src_addr):
- """
- Verify IPFIX maximum IPv6 fragments pending reassembly exceeded event
-
- :param data: Decoded IPFIX data records
- :param limit: Number of maximum fragments pending reassembly
- :param src_addr: IPv6 source address
- """
- self.assertEqual(1, len(data))
- record = data[0]
- # natEvent
- self.assertEqual(scapy.compat.orb(record[230]), 13)
- # natQuotaExceededEvent
- self.assertEqual(struct.pack("I", 5), record[466])
- # maxFragmentsPendingReassembly
- self.assertEqual(struct.pack("I", limit), record[475])
- # sourceIPv6Address
- self.assertEqual(src_addr, record[27])
-
- def verify_ipfix_max_fragments_ip4(self, data, limit, src_addr):
- """
- Verify IPFIX maximum IPv4 fragments pending reassembly exceeded event
-
- :param data: Decoded IPFIX data records
- :param limit: Number of maximum fragments pending reassembly
- :param src_addr: IPv4 source address
- """
- self.assertEqual(1, len(data))
- record = data[0]
- # natEvent
- self.assertEqual(scapy.compat.orb(record[230]), 13)
- # natQuotaExceededEvent
- self.assertEqual(struct.pack("I", 5), record[466])
- # maxFragmentsPendingReassembly
- self.assertEqual(struct.pack("I", limit), record[475])
- # sourceIPv4Address
- self.assertEqual(src_addr, record[8])
-
def verify_ipfix_bib(self, data, is_create, src_addr):
"""
Verify IPFIX NAT64 BIB create and delete events
else:
self.assertEqual(scapy.compat.orb(record[230]), 11)
# sourceIPv6Address
- self.assertEqual(src_addr, record[27])
+ self.assertEqual(src_addr, str(ipaddress.IPv6Address(record[27])))
# postNATSourceIPv4Address
self.assertEqual(self.nat_addr_n, record[225])
# protocolIdentifier
else:
self.assertEqual(scapy.compat.orb(record[230]), 7)
# sourceIPv6Address
- self.assertEqual(src_addr, record[27])
+ self.assertEqual(src_addr, str(ipaddress.IPv6Address(record[27])))
# destinationIPv6Address
self.assertEqual(socket.inet_pton(socket.AF_INET6,
self.compose_ip6(dst_addr,
else:
raise Exception("Unsupported protocol")
- def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False):
+ def frag_in_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
+ ignore_port=False):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
if proto != IP_PROTOS.icmp:
if not dont_translate:
self.assertEqual(p[layer].dport, 20)
- self.assertNotEqual(p[layer].sport, self.port_in)
+ if not ignore_port:
+ self.assertNotEqual(p[layer].sport, self.port_in)
else:
self.assertEqual(p[layer].sport, self.port_in)
else:
- if not dont_translate:
- self.assertNotEqual(p[layer].id, self.port_in)
- else:
- self.assertEqual(p[layer].id, self.port_in)
+ if not ignore_port:
+ if not dont_translate:
+ self.assertNotEqual(p[layer].id, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
self.assertEqual(data, p[Raw].load)
# out2in
self.assertEqual(p[layer].id, self.port_in)
self.assertEqual(data, p[Raw].load)
- def reass_hairpinning(self, proto=IP_PROTOS.tcp):
+ def reass_hairpinning(self, proto=IP_PROTOS.tcp, ignore_port=False):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
self.nat_addr,
self.server.ip4)
if proto != IP_PROTOS.icmp:
- self.assertNotEqual(p[layer].sport, self.host_in_port)
+ if not ignore_port:
+ self.assertNotEqual(p[layer].sport, self.host_in_port)
self.assertEqual(p[layer].dport, self.server_in_port)
else:
- self.assertNotEqual(p[layer].id, self.host_in_port)
+ if not ignore_port:
+ self.assertNotEqual(p[layer].id, self.host_in_port)
self.assertEqual(data, p[Raw].load)
- def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False):
+ def frag_out_of_order(self, proto=IP_PROTOS.tcp, dont_translate=False,
+ ignore_port=False):
layer = self.proto2layer(proto)
if proto == IP_PROTOS.tcp:
if proto != IP_PROTOS.icmp:
if not dont_translate:
self.assertEqual(p[layer].dport, 20)
- self.assertNotEqual(p[layer].sport, self.port_in)
+ if not ignore_port:
+ self.assertNotEqual(p[layer].sport, self.port_in)
else:
self.assertEqual(p[layer].sport, self.port_in)
else:
- if not dont_translate:
- self.assertNotEqual(p[layer].id, self.port_in)
- else:
- self.assertEqual(p[layer].id, self.port_in)
+ if not ignore_port:
+ if not dont_translate:
+ self.assertNotEqual(p[layer].id, self.port_in)
+ else:
+ self.assertEqual(p[layer].id, self.port_in)
self.assertEqual(data, p[Raw].load)
# out2in
class TestNAT44(MethodHolder):
""" NAT44 Test Cases """
+ max_translations = 10240
+ max_users = 10240
+
+ @classmethod
+ def setUpConstants(cls):
+ super(TestNAT44, cls).setUpConstants()
+ cls.vpp_cmdline.extend([
+ "nat", "{",
+ "max translations per thread %d" % cls.max_translations,
+ "max users per thread %d" % cls.max_users,
+ "}"
+ ])
+
@classmethod
def setUpClass(cls):
super(TestNAT44, cls).setUpClass()
def tearDownClass(cls):
super(TestNAT44, cls).tearDownClass()
+ def test_clear_sessions(self):
+ """ NAT44 session clearing test """
+
+ self.nat44_add_address(self.nat_addr)
+ flags = self.config_flags.NAT_IS_INSIDE
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(0, nat_config.endpoint_dependent)
+
+ 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)
+
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertTrue(sessions[0][0] > 0)
+ self.logger.info("sessions before clearing: %s" % sessions[0][0])
+
+ self.vapi.cli("clear nat44 sessions")
+
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 0)
+ self.logger.info("sessions after clearing: %s" % sessions[0][0])
+
def test_dynamic(self):
""" NAT44 dynamic translation test """
self.nat44_add_address(self.nat_addr)
is_add=1)
# in2out
- tcpn = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/TCP packets')
- udpn = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/good in2out packets processed')
+ tcpn = self.statistics.get_counter('/nat44/in2out/slowpath/tcp')[0]
+ udpn = self.statistics.get_counter('/nat44/in2out/slowpath/udp')[0]
+ icmpn = self.statistics.get_counter('/nat44/in2out/slowpath/icmp')[0]
+ drops = self.statistics.get_counter('/nat44/in2out/slowpath/drops')[0]
pkts = self.create_stream_in(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
capture = self.pg1.get_capture(len(pkts))
self.verify_capture_out(capture)
- err = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-in2out-slowpath/good in2out packets processed')
- self.assertEqual(err - totaln, 4)
+ if_idx = self.pg0.sw_if_index
+ cnt = self.statistics.get_counter('/nat44/in2out/slowpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter('/nat44/in2out/slowpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/in2out/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/in2out/slowpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
# out2in
- tcpn = self.statistics.get_err_counter('/err/nat44-out2in/TCP packets')
- udpn = self.statistics.get_err_counter('/err/nat44-out2in/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-out2in/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-out2in/good out2in packets processed')
+ tcpn = self.statistics.get_counter('/nat44/out2in/slowpath/tcp')[0]
+ udpn = self.statistics.get_counter('/nat44/out2in/slowpath/udp')[0]
+ icmpn = self.statistics.get_counter('/nat44/out2in/slowpath/icmp')[0]
+ drops = self.statistics.get_counter('/nat44/out2in/slowpath/drops')[0]
pkts = self.create_stream_out(self.pg1)
self.pg1.add_stream(pkts)
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
- err = self.statistics.get_err_counter('/err/nat44-out2in/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter('/err/nat44-out2in/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter('/err/nat44-out2in/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-out2in/good out2in packets processed')
- self.assertEqual(err - totaln, 4)
+ if_idx = self.pg1.sw_if_index
+ cnt = self.statistics.get_counter('/nat44/out2in/slowpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter('/nat44/out2in/slowpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/out2in/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/out2in/slowpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
users = self.statistics.get_counter('/nat44/total-users')
self.assertEqual(users[0][0], 1)
server_in_port, server_out_port,
proto=IP_PROTOS.tcp)
+ cnt = self.statistics.get_counter('/nat44/hairpinning')[0]
# send packet from host to server
p = (Ether(src=host.mac, dst=self.pg0.local_mac) /
IP(src=host.ip4, dst=self.nat_addr) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ if_idx = self.pg0.sw_if_index
+ self.assertEqual(after[if_idx] - cnt[if_idx], 1)
+
# send reply from server to host
p = (Ether(src=server.mac, dst=self.pg0.local_mac) /
IP(src=server.ip4, dst=self.nat_addr) /
self.logger.error(ppp("Unexpected or invalid packet:", p))
raise
+ after = self.statistics.get_counter('/nat44/hairpinning')[0]
+ if_idx = self.pg0.sw_if_index
+ self.assertEqual(after[if_idx] - cnt[if_idx], 2)
+
def test_hairpinning2(self):
""" NAT44 hairpinning - 1:1 NAT"""
self.verify_capture_out(capture)
self.nat44_add_address(self.nat_addr, is_add=0)
self.vapi.ipfix_flush()
- capture = self.pg3.get_capture(9)
+ capture = self.pg3.get_capture(7)
ipfix = IPFIXDecoder()
# first load template
for p in capture:
self.pg1.assert_nothing_captured()
sleep(1)
self.vapi.ipfix_flush()
- capture = self.pg3.get_capture(9)
+ capture = self.pg3.get_capture(7)
ipfix = IPFIXDecoder()
# first load template
for p in capture:
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- nat44_config = self.vapi.nat_show_config()
- max_sessions = 10 * nat44_config.translation_buckets
+ max_sessions = self.max_translations
pkts = []
for i in range(0, max_sessions):
self.pg1.assert_nothing_captured()
sleep(1)
self.vapi.ipfix_flush()
- capture = self.pg3.get_capture(9)
+ capture = self.pg3.get_capture(7)
ipfix = IPFIXDecoder()
# first load template
for p in capture:
class TestNAT44EndpointDependent2(MethodHolder):
- """ Endpoint-Dependent session test cases """
-
- icmp_timeout = 2
+ """ Endpoint-Dependent mapping and filtering test cases """
@classmethod
def setUpConstants(cls):
super(TestNAT44EndpointDependent2, cls).setUpConstants()
- cls.vpp_cmdline.extend(["nat", "{", "endpoint-dependent",
- "translation", "hash", "buckets", "1",
- "icmp", "timeout", str(cls.icmp_timeout), "}"])
+ cls.vpp_cmdline.extend(["nat", "{", "endpoint-dependent", "}"])
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT44EndpointDependent2, cls).tearDownClass()
+
+ def tearDown(self):
+ super(TestNAT44EndpointDependent2, self).tearDown()
+
+ @classmethod
+ def create_and_add_ip4_table(cls, i, table_id):
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': table_id})
+ i.set_table_ip4(table_id)
@classmethod
def setUpClass(cls):
super(TestNAT44EndpointDependent2, cls).setUpClass()
- translation_buckets = 1
- cls.max_translations = 10 * translation_buckets
- cls.create_pg_interfaces(range(2))
- cls.interfaces = list(cls.pg_interfaces[0:2])
+ cls.create_pg_interfaces(range(3))
+ cls.interfaces = list(cls.pg_interfaces)
+
+ cls.create_and_add_ip4_table(cls.pg1, 10)
for i in cls.interfaces:
i.admin_up()
i.config_ip4()
i.resolve_arp()
- cls.pg0.generate_remote_hosts(1)
- cls.pg0.configure_ipv4_neighbors()
+ i.generate_remote_hosts(1)
+ i.configure_ipv4_neighbors()
- cls.pg1.generate_remote_hosts(1)
- cls.pg1.configure_ipv4_neighbors()
+ def setUp(self):
+ super(TestNAT44EndpointDependent2, self).setUp()
- @classmethod
- def tearDownClass(cls):
- super(TestNAT44EndpointDependent2, cls).tearDownClass()
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(1, nat_config.endpoint_dependent)
+
+ def nat_add_inside_interface(self, i):
+ self.vapi.nat44_interface_add_del_feature(
+ flags=self.config_flags.NAT_IS_INSIDE,
+ sw_if_index=i.sw_if_index, is_add=1)
+
+ def nat_add_outside_interface(self, i):
+ self.vapi.nat44_interface_add_del_feature(
+ flags=self.config_flags.NAT_IS_OUTSIDE,
+ sw_if_index=i.sw_if_index, is_add=1)
+
+ def nat_add_interface_address(self, i):
+ self.nat_addr = i.local_ip4
+ self.vapi.nat44_add_del_interface_addr(
+ sw_if_index=i.sw_if_index, is_add=1)
+
+ def nat_add_address(self, address, vrf_id=0xFFFFFFFF):
+ self.nat_addr = address
+ self.nat44_add_address(address, vrf_id=vrf_id)
- def create_icmp_stream(self, in_if, out_if, count):
+ def cli(self, command):
+ result = self.vapi.cli(command)
+ self.logger.info(result)
+ # print(result)
+
+ def show_configuration(self):
+ self.cli("show interface")
+ self.cli("show interface address")
+ self.cli("show nat44 addresses")
+ self.cli("show nat44 interfaces")
+
+ def create_tcp_stream(self, in_if, out_if, count):
"""
- Create ICMP packet stream for inside network
+ Create tcp packet stream
:param in_if: Inside interface
:param out_if: Outside interface
- :param count: Number of packets
+ :param count: count of packets to generate
"""
+ pkts = []
+ port = 6303
- self.assertTrue(count > 0)
- icmp_id = random.randint(0, 65535 - (count - 1))
-
- pkts = list()
for i in range(count):
p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) /
IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=64) /
- ICMP(id=icmp_id + i, type='echo-request'))
+ TCP(sport=port + i, dport=20))
pkts.append(p)
+
return pkts
- def send_pkts(self, pkts, expected=None):
- self.pg0.add_stream(pkts)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- return self.pg1.get_capture(
- len(pkts) if expected is None else expected)
+ def test_session_limit_per_vrf(self):
- def test_session_cleanup(self):
- """ NAT44 session cleanup test """
+ inside = self.pg0
+ inside_vrf10 = self.pg1
+ outside = self.pg2
- self.nat44_add_address(self.pg1.local_ip4)
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat44_interface_add_del_feature(
- sw_if_index=self.pg0.sw_if_index,
- flags=flags, is_add=1)
- self.vapi.nat44_interface_add_del_feature(
- sw_if_index=self.pg1.sw_if_index,
- is_add=1)
+ limit = 5
- nat_config = self.vapi.nat_show_config()
- self.assertEqual(1, nat_config.endpoint_dependent)
+ # 2 interfaces pg0, pg1 (vrf10, limit 1 tcp session)
+ # non existing vrf_id makes process core dump
+ self.vapi.nat44_set_session_limit(session_limit=limit, vrf_id=10)
+
+ self.nat_add_inside_interface(inside)
+ self.nat_add_inside_interface(inside_vrf10)
+ self.nat_add_outside_interface(outside)
- pkts = self.create_icmp_stream(self.pg0, self.pg1,
- self.max_translations + 2)
- sz = len(pkts)
+ # vrf independent
+ self.nat_add_interface_address(outside)
- # positive test
- self.send_pkts(pkts[0:self.max_translations])
+ # BUG: causing core dump - when bad vrf_id is specified
+ # self.nat44_add_address(outside.local_ip4, vrf_id=20)
- # false positive test
- self.send_pkts(pkts[self.max_translations:sz - 1], 0)
+ self.show_configuration()
- sleep(self.icmp_timeout)
+ stream = self.create_tcp_stream(inside_vrf10, outside, limit * 2)
+ inside_vrf10.add_stream(stream)
- # positive test
- self.send_pkts(pkts[self.max_translations + 1:sz])
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ capture = outside.get_capture(limit)
+
+ stream = self.create_tcp_stream(inside, outside, limit * 2)
+ inside.add_stream(stream)
+
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+
+ capture = outside.get_capture(len(stream))
class TestNAT44EndpointDependent(MethodHolder):
self.vapi.nat44_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.frag_in_order(proto=IP_PROTOS.tcp)
- self.frag_in_order(proto=IP_PROTOS.udp)
- self.frag_in_order(proto=IP_PROTOS.icmp)
+ self.frag_in_order(proto=IP_PROTOS.tcp, ignore_port=True)
+ self.frag_in_order(proto=IP_PROTOS.udp, ignore_port=True)
+ self.frag_in_order(proto=IP_PROTOS.icmp, ignore_port=True)
def test_frag_in_order_dont_translate(self):
""" NAT44 don't translate fragments arriving in order """
self.vapi.nat44_interface_add_del_feature(
sw_if_index=self.pg1.sw_if_index,
is_add=1)
- self.frag_out_of_order(proto=IP_PROTOS.tcp)
- self.frag_out_of_order(proto=IP_PROTOS.udp)
- self.frag_out_of_order(proto=IP_PROTOS.icmp)
+ self.frag_out_of_order(proto=IP_PROTOS.tcp, ignore_port=True)
+ self.frag_out_of_order(proto=IP_PROTOS.udp, ignore_port=True)
+ self.frag_out_of_order(proto=IP_PROTOS.icmp, ignore_port=True)
def test_frag_out_of_order_dont_translate(self):
""" NAT44 don't translate fragments arriving out of order """
proto=IP_PROTOS.udp)
self.nat44_add_static_mapping(self.server.ip4, self.nat_addr)
- self.reass_hairpinning(proto=IP_PROTOS.tcp)
- self.reass_hairpinning(proto=IP_PROTOS.udp)
- self.reass_hairpinning(proto=IP_PROTOS.icmp)
+ self.reass_hairpinning(proto=IP_PROTOS.tcp, ignore_port=True)
+ self.reass_hairpinning(proto=IP_PROTOS.udp, ignore_port=True)
+ self.reass_hairpinning(proto=IP_PROTOS.icmp, ignore_port=True)
+
+ def test_clear_sessions(self):
+ """ NAT44 ED session clearing test """
+
+ self.nat44_add_address(self.nat_addr)
+ flags = self.config_flags.NAT_IS_INSIDE
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(1, nat_config.endpoint_dependent)
+
+ 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, ignore_port=True)
+
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertTrue(sessions[0][0] > 0)
+ self.logger.info("sessions before clearing: %s" % sessions[0][0])
+
+ # just for testing purposes
+ self.logger.info(self.vapi.cli("show nat44 summary"))
+
+ self.vapi.cli("clear nat44 sessions")
+
+ self.logger.info(self.vapi.cli("show nat44 summary"))
+
+ sessions = self.statistics.get_counter('/nat44/total-sessions')
+ self.assertEqual(sessions[0][0], 0)
+ self.logger.info("sessions after clearing: %s" % sessions[0][0])
def test_dynamic(self):
""" NAT44 dynamic translation test """
self.assertEqual(1, nat_config.endpoint_dependent)
# in2out
- tcpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/TCP packets')
- udpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/good in2out packets processed')
+ tcpn = self.statistics.get_counter('/nat44/ed/in2out/slowpath/tcp')[0]
+ udpn = self.statistics.get_counter('/nat44/ed/in2out/slowpath/udp')[0]
+ icmpn = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/icmp')[0]
+ drops = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/drops')[0]
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)
-
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/good in2out packets processed')
- self.assertEqual(err - totaln, 4)
+ self.verify_capture_out(capture, ignore_port=True)
+
+ if_idx = self.pg0.sw_if_index
+ cnt = self.statistics.get_counter('/nat44/ed/in2out/slowpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter('/nat44/ed/in2out/slowpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/ed/in2out/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/ed/in2out/slowpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
# out2in
- tcpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/TCP packets')
- udpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in-slowpath/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/good out2in packets processed')
+ tcpn = self.statistics.get_counter('/nat44/ed/out2in/fastpath/tcp')[0]
+ udpn = self.statistics.get_counter('/nat44/ed/out2in/fastpath/udp')[0]
+ icmpn = self.statistics.get_counter(
+ '/nat44/ed/out2in/slowpath/icmp')[0]
+ drops = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/drops')[0]
pkts = self.create_stream_out(self.pg1)
self.pg1.add_stream(pkts)
capture = self.pg0.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg0)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in-slowpath/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/good out2in packets processed')
- self.assertEqual(err - totaln, 3)
+ if_idx = self.pg1.sw_if_index
+ cnt = self.statistics.get_counter('/nat44/ed/out2in/fastpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter('/nat44/ed/out2in/fastpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/ed/out2in/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat44/ed/out2in/fastpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
- users = self.statistics.get_counter('/nat44/total-users')
- self.assertEqual(users[0][0], 1)
sessions = self.statistics.get_counter('/nat44/total-sessions')
self.assertEqual(sessions[0][0], 3)
+ def test_dynamic_out_of_ports(self):
+ """ NAT44 dynamic translation test: out of ports """
+
+ flags = self.config_flags.NAT_IS_INSIDE
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(1, nat_config.endpoint_dependent)
+
+ # in2out and no NAT addresses added
+ err_old = self.statistics.get_err_counter(
+ '/err/nat44-ed-in2out-slowpath/out of ports')
+
+ pkts = self.create_stream_in(self.pg0, self.pg1)
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(0, timeout=1)
+
+ err_new = self.statistics.get_err_counter(
+ '/err/nat44-ed-in2out-slowpath/out of ports')
+
+ self.assertEqual(err_new - err_old, len(pkts))
+
+ # in2out after NAT addresses added
+ self.nat44_add_address(self.nat_addr)
+
+ err_old = self.statistics.get_err_counter(
+ '/err/nat44-ed-in2out-slowpath/out of ports')
+
+ 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, ignore_port=True)
+
+ err_new = self.statistics.get_err_counter(
+ '/err/nat44-ed-in2out-slowpath/out of ports')
+
+ self.assertEqual(err_new, err_old)
+
def test_dynamic_output_feature_vrf(self):
""" NAT44 dynamic translation test: output-feature, VRF"""
self.assertEqual(1, nat_config.endpoint_dependent)
# in2out
- tcpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/TCP packets')
- udpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/good in2out packets processed')
+ tcpn = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/tcp')[0]
+ udpn = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/udp')[0]
+ icmpn = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/icmp')[0]
+ drops = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/drops')[0]
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)
-
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-in2out-slowpath/good in2out packets processed')
- self.assertEqual(err - totaln, 4)
+ self.verify_capture_out(capture, ignore_port=True)
+
+ if_idx = self.pg7.sw_if_index
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/in2out/slowpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
# out2in
- tcpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/TCP packets')
- udpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in-slowpath/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/good out2in packets processed')
+ tcpn = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/tcp')[0]
+ udpn = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/udp')[0]
+ icmpn = self.statistics.get_counter(
+ '/nat44/ed/out2in/slowpath/icmp')[0]
+ drops = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/drops')[0]
pkts = self.create_stream_out(self.pg8)
self.pg8.add_stream(pkts)
capture = self.pg7.get_capture(len(pkts))
self.verify_capture_in(capture, self.pg7)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in-slowpath/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat44-ed-out2in/good out2in packets processed')
- self.assertEqual(err - totaln, 3)
-
- users = self.statistics.get_counter('/nat44/total-users')
- self.assertEqual(users[0][0], 1)
+ if_idx = self.pg8.sw_if_index
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/out2in/slowpath/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter(
+ '/nat44/ed/out2in/fastpath/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
+
sessions = self.statistics.get_counter('/nat44/total-sessions')
self.assertEqual(sessions[0][0], 3)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
+ self.verify_capture_out(capture, ignore_port=True)
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.verify_capture_out(capture, ignore_port=True)
# from external network back to local network host
pkts = self.create_stream_out(self.pg1)
self.pg_enable_capture(self.pg_interfaces)
self.pg_start()
capture = self.pg1.get_capture(len(pkts))
- self.verify_capture_out(capture)
+ self.verify_capture_out(capture, ignore_port=True)
pkts = self.create_stream_out(self.pg1)
self.pg1.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.verify_capture_out(capture, ignore_port=True)
# out2in
pkts = self.create_stream_out(self.pg1)
pkts_in2out = self.create_stream_in(self.pg0, self.pg1)
capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
len(pkts_in2out))
- self.verify_capture_out(capture)
+ self.verify_capture_out(capture, ignore_port=True)
# send out2in again, with sessions created it should work now
pkts_out2in = self.create_stream_out(self.pg1)
# send in2out to generate ACL state (NAT state was created earlier)
capture = self.send_and_expect(self.pg0, pkts_in2out, self.pg1,
len(pkts_in2out))
- self.verify_capture_out(capture)
+ self.verify_capture_out(capture, ignore_port=True)
# send out2in again. ACL state exists so it should work now.
# TCP packets with the syn flag set also need the ack flag
ip = p[IP]
tcp = p[TCP]
self.assertEqual(ip.src, self.nat_addr)
- self.assertNotEqual(tcp.sport, 2345)
self.assert_packet_checksums_valid(p)
port = tcp.sport
except:
self.pg_start()
self.pg1.get_capture(1)
- nsessions = 0
- users = self.vapi.nat44_user_dump()
- self.assertEqual(len(users), 1)
- self.assertEqual(str(users[0].ip_address),
- self.pg0.remote_ip4)
- self.assertEqual(users[0].nsessions, 1)
-
def test_syslog_sess(self):
""" Test syslog session creation and deletion """
self.vapi.syslog_set_filter(
capture = self.pg2.get_capture(1)
self.verify_syslog_sess(capture[0][Raw].load, False)
+ def test_ed_users_dump(self):
+ """ API test - nat44_user_dump """
+ flags = self.config_flags.NAT_IS_INSIDE
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index,
+ flags=flags, is_add=1)
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index,
+ is_add=1)
+ self.vapi.nat44_forwarding_enable_disable(enable=1)
+
+ real_ip = self.pg0.remote_ip4
+ alias_ip = self.nat_addr
+ flags = self.config_flags.NAT_IS_ADDR_ONLY
+ self.vapi.nat44_add_del_static_mapping(is_add=1,
+ local_ip_address=real_ip,
+ external_ip_address=alias_ip,
+ external_sw_if_index=0xFFFFFFFF,
+ flags=flags)
+
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 0)
+ try:
+ # in2out - static mapping match
+
+ 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)
+
+ 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, same_port=True)
+
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 1)
+ static_user = users[0]
+ self.assertEqual(static_user.nstaticsessions, 3)
+ self.assertEqual(static_user.nsessions, 0)
+
+ # in2out - no static mapping match
+
+ host0 = self.pg0.remote_hosts[0]
+ self.pg0.remote_hosts[0] = self.pg0.remote_hosts[1]
+ try:
+ pkts = self.create_stream_out(self.pg1,
+ dst_ip=self.pg0.remote_ip4,
+ use_inside_ports=True)
+ 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)
+
+ 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=self.pg0.remote_ip4,
+ same_port=True)
+ finally:
+ self.pg0.remote_hosts[0] = host0
+
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 2)
+ if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
+ non_static_user = users[1]
+ static_user = users[0]
+ else:
+ non_static_user = users[0]
+ static_user = users[1]
+ self.assertEqual(static_user.nstaticsessions, 3)
+ self.assertEqual(static_user.nsessions, 0)
+ self.assertEqual(non_static_user.nstaticsessions, 0)
+ self.assertEqual(non_static_user.nsessions, 3)
+
+ users = self.vapi.nat44_user_dump()
+ self.assertEqual(len(users), 2)
+ if str(users[0].ip_address) == self.pg0.remote_hosts[0].ip4:
+ non_static_user = users[1]
+ static_user = users[0]
+ else:
+ non_static_user = users[0]
+ static_user = users[1]
+ self.assertEqual(static_user.nstaticsessions, 3)
+ self.assertEqual(static_user.nsessions, 0)
+ self.assertEqual(non_static_user.nstaticsessions, 0)
+ self.assertEqual(non_static_user.nsessions, 3)
+
+ finally:
+ self.vapi.nat44_forwarding_enable_disable(enable=0)
+ flags = self.config_flags.NAT_IS_ADDR_ONLY
+ self.vapi.nat44_add_del_static_mapping(
+ is_add=0,
+ local_ip_address=real_ip,
+ external_ip_address=alias_ip,
+ external_sw_if_index=0xFFFFFFFF,
+ flags=flags)
+
def tearDown(self):
super(TestNAT44EndpointDependent, self).tearDown()
if not self.vpp_dead:
self.logger.info(self.vapi.cli("show nat timeouts"))
+class TestNAT44EndpointDependent3(MethodHolder):
+ """ Endpoint-Dependent mapping and filtering extra test cases """
+
+ max_translations = 50
+
+ @classmethod
+ def setUpConstants(cls):
+ super(TestNAT44EndpointDependent3, cls).setUpConstants()
+ cls.vpp_cmdline.extend([
+ "nat", "{", "endpoint-dependent",
+ "max translations per thread %d" % cls.max_translations,
+ "}"
+ ])
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestNAT44EndpointDependent3, cls).setUpClass()
+ cls.vapi.cli("set log class nat level debug")
+
+ cls.nat_addr = '10.0.0.3'
+
+ cls.create_pg_interfaces(range(2))
+
+ for i in cls.pg_interfaces:
+ i.admin_up()
+ i.config_ip4()
+ i.resolve_arp()
+
+ def setUp(self):
+ super(TestNAT44EndpointDependent3, self).setUp()
+ self.vapi.nat_set_timeouts(
+ udp=1, tcp_established=7440, tcp_transitory=30, icmp=1)
+ self.nat44_add_address(self.nat_addr)
+ flags = self.config_flags.NAT_IS_INSIDE
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg0.sw_if_index, flags=flags, is_add=1)
+ self.vapi.nat44_interface_add_del_feature(
+ sw_if_index=self.pg1.sw_if_index, is_add=1)
+
+ @classmethod
+ def tearDownClass(cls):
+ super(TestNAT44EndpointDependent3, cls).tearDownClass()
+
+ def init_tcp_session(self, in_if, out_if, sport, ext_dport):
+ # 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=sport, dport=ext_dport, 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]
+ 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.nat_addr) /
+ TCP(sport=ext_dport, dport=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=sport, dport=ext_dport, flags="A"))
+ in_if.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ out_if.get_capture(1)
+
+ return tcp_port_out
+
+ def test_lru_cleanup(self):
+ """ LRU cleanup algorithm """
+ tcp_port_out = self.init_tcp_session(self.pg0, self.pg1, 2000, 80)
+ pkts = []
+ for i in range(0, self.max_translations - 1):
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(src=self.pg0.remote_ip4, dst=self.pg1.remote_ip4, ttl=64) /
+ UDP(sport=7000+i, dport=80))
+ pkts.append(p)
+
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(len(pkts))
+ self.sleep(1.5, "wait for timeouts")
+
+ pkts = []
+ for i in range(0, self.max_translations - 1):
+ p = (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=8000+i, type='echo-request'))
+ pkts.append(p)
+
+ self.pg0.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ self.pg1.get_capture(len(pkts))
+
+
class TestNAT44Out2InDPO(MethodHolder):
""" NAT44 Test Cases using out2in DPO """
sw_if_index=self.pg1.sw_if_index)
# in2out
- tcpn = self.statistics.get_err_counter('/err/nat64-in2out/TCP packets')
- udpn = self.statistics.get_err_counter('/err/nat64-in2out/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat64-in2out/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat64-in2out/good in2out packets processed')
+ tcpn = self.statistics.get_counter('/nat64/in2out/tcp')[0]
+ udpn = self.statistics.get_counter('/nat64/in2out/udp')[0]
+ icmpn = self.statistics.get_counter('/nat64/in2out/icmp')[0]
+ drops = self.statistics.get_counter('/nat64/in2out/drops')[0]
pkts = self.create_stream_in_ip6(self.pg0, self.pg1)
self.pg0.add_stream(pkts)
self.verify_capture_out(capture, nat_ip=self.nat_addr,
dst_ip=self.pg1.remote_ip4)
- err = self.statistics.get_err_counter('/err/nat64-in2out/TCP packets')
- self.assertEqual(err - tcpn, 1)
- err = self.statistics.get_err_counter('/err/nat64-in2out/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter('/err/nat64-in2out/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat64-in2out/good in2out packets processed')
- self.assertEqual(err - totaln, 3)
+ if_idx = self.pg0.sw_if_index
+ cnt = self.statistics.get_counter('/nat64/in2out/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat64/in2out/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat64/in2out/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat64/in2out/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
# out2in
- tcpn = self.statistics.get_err_counter('/err/nat64-out2in/TCP packets')
- udpn = self.statistics.get_err_counter('/err/nat64-out2in/UDP packets')
- icmpn = self.statistics.get_err_counter(
- '/err/nat64-out2in/ICMP packets')
- totaln = self.statistics.get_err_counter(
- '/err/nat64-out2in/good out2in packets processed')
+ tcpn = self.statistics.get_counter('/nat64/out2in/tcp')[0]
+ udpn = self.statistics.get_counter('/nat64/out2in/udp')[0]
+ icmpn = self.statistics.get_counter('/nat64/out2in/icmp')[0]
+ drops = self.statistics.get_counter('/nat64/out2in/drops')[0]
pkts = self.create_stream_out(self.pg1, dst_ip=self.nat_addr)
self.pg1.add_stream(pkts)
ip = IPv6(src=''.join(['64:ff9b::', self.pg1.remote_ip4]))
self.verify_capture_in_ip6(capture, ip[IPv6].src, self.pg0.remote_ip6)
- err = self.statistics.get_err_counter('/err/nat64-out2in/TCP packets')
- self.assertEqual(err - tcpn, 2)
- err = self.statistics.get_err_counter('/err/nat64-out2in/UDP packets')
- self.assertEqual(err - udpn, 1)
- err = self.statistics.get_err_counter('/err/nat64-out2in/ICMP packets')
- self.assertEqual(err - icmpn, 1)
- err = self.statistics.get_err_counter(
- '/err/nat64-out2in/good out2in packets processed')
- self.assertEqual(err - totaln, 4)
+ if_idx = self.pg1.sw_if_index
+ cnt = self.statistics.get_counter('/nat64/out2in/tcp')[0]
+ self.assertEqual(cnt[if_idx] - tcpn[if_idx], 2)
+ cnt = self.statistics.get_counter('/nat64/out2in/udp')[0]
+ self.assertEqual(cnt[if_idx] - udpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat64/out2in/icmp')[0]
+ self.assertEqual(cnt[if_idx] - icmpn[if_idx], 1)
+ cnt = self.statistics.get_counter('/nat64/out2in/drops')[0]
+ self.assertEqual(cnt[if_idx] - drops[if_idx], 0)
bibs = self.statistics.get_counter('/nat64/total-bibs')
self.assertEqual(bibs[0][0], 3)
self.vapi.nat64_add_del_interface(is_add=1, flags=0,
sw_if_index=self.pg1.sw_if_index)
- self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6n,
+ self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6,
o_addr=self.nat_addr,
i_port=self.tcp_port_in,
o_port=self.tcp_port_out,
proto=IP_PROTOS.tcp, vrf_id=0,
is_add=1)
- self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6n,
+ self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6,
o_addr=self.nat_addr,
i_port=self.udp_port_in,
o_port=self.udp_port_out,
proto=IP_PROTOS.udp, vrf_id=0,
is_add=1)
- self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6n,
+ self.vapi.nat64_add_del_static_bib(i_addr=self.pg0.remote_ip6,
o_addr=self.nat_addr,
i_port=self.icmp_id_in,
o_port=self.icmp_id_out,
self.pg1.assert_nothing_captured()
sleep(1)
self.vapi.ipfix_flush()
- capture = self.pg3.get_capture(9)
+ capture = self.pg3.get_capture(7)
ipfix = IPFIXDecoder()
# first load template
for p in capture:
p = self.pg1.get_capture(1)
self.tcp_port_out = p[0][TCP].sport
self.vapi.ipfix_flush()
- capture = self.pg3.get_capture(10)
+ capture = self.pg3.get_capture(8)
ipfix = IPFIXDecoder()
# first load template
for p in capture:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
if scapy.compat.orb(data[0][230]) == 10:
- self.verify_ipfix_bib(data, 1, self.pg0.remote_ip6n)
+ self.verify_ipfix_bib(data, 1, self.pg0.remote_ip6)
elif scapy.compat.orb(data[0][230]) == 6:
self.verify_ipfix_nat64_ses(data,
1,
- self.pg0.remote_ip6n,
+ self.pg0.remote_ip6,
self.pg1.remote_ip4,
25)
else:
if p.haslayer(Data):
data = ipfix.decode_data_set(p.getlayer(Set))
if scapy.compat.orb(data[0][230]) == 11:
- self.verify_ipfix_bib(data, 0, self.pg0.remote_ip6n)
+ self.verify_ipfix_bib(data, 0, self.pg0.remote_ip6)
elif scapy.compat.orb(data[0][230]) == 7:
self.verify_ipfix_nat64_ses(data,
0,
- self.pg0.remote_ip6n,
+ self.pg0.remote_ip6,
self.pg1.remote_ip4,
25)
else:
self.logger.info(self.vapi.cli("show nat64 session table all"))
-class TestNAT66(MethodHolder):
- """ NAT66 Test Cases """
-
- @classmethod
- def setUpClass(cls):
- super(TestNAT66, cls).setUpClass()
-
- cls.nat_addr = 'fd01:ff::2'
-
- cls.create_pg_interfaces(range(2))
- cls.interfaces = list(cls.pg_interfaces)
-
- for i in cls.interfaces:
- i.admin_up()
- i.config_ip6()
- i.configure_ipv6_neighbors()
-
- @classmethod
- def tearDownClass(cls):
- super(TestNAT66, cls).tearDownClass()
-
- def test_static(self):
- """ 1:1 NAT66 test """
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
- sw_if_index=self.pg0.sw_if_index)
- self.vapi.nat66_add_del_interface(is_add=1,
- sw_if_index=self.pg1.sw_if_index)
- self.vapi.nat66_add_del_static_mapping(
- local_ip_address=self.pg0.remote_ip6n,
- external_ip_address=self.nat_addr,
- is_add=1)
-
- # in2out
- pkts = []
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
- TCP())
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
- UDP())
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
- ICMPv6EchoRequest())
- pkts.append(p)
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
- GRE() / IP() / TCP())
- 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))
-
- for packet in capture:
- try:
- self.assertEqual(packet[IPv6].src, self.nat_addr)
- self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
- self.assert_packet_checksums_valid(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- # out2in
- pkts = []
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
- TCP())
- pkts.append(p)
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
- UDP())
- pkts.append(p)
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
- ICMPv6EchoReply())
- pkts.append(p)
- p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
- IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
- GRE() / IP() / TCP())
- pkts.append(p)
- 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, self.pg1.remote_ip6)
- self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
- self.assert_packet_checksums_valid(packet)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- sm = self.vapi.nat66_static_mapping_dump()
- self.assertEqual(len(sm), 1)
- self.assertEqual(sm[0].total_pkts, 8)
-
- def test_check_no_translate(self):
- """ NAT66 translate only when egress interface is outside interface """
- flags = self.config_flags.NAT_IS_INSIDE
- self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
- sw_if_index=self.pg0.sw_if_index)
- self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
- sw_if_index=self.pg1.sw_if_index)
- self.vapi.nat66_add_del_static_mapping(
- local_ip_address=self.pg0.remote_ip6n,
- external_ip_address=self.nat_addr,
- is_add=1)
-
- # in2out
- p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
- IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
- UDP())
- self.pg0.add_stream([p])
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- capture = self.pg1.get_capture(1)
- packet = capture[0]
- try:
- self.assertEqual(packet[IPv6].src, self.pg0.remote_ip6)
- self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
- except:
- self.logger.error(ppp("Unexpected or invalid packet:", packet))
- raise
-
- def clear_nat66(self):
- """
- Clear NAT66 configuration.
- """
- interfaces = self.vapi.nat66_interface_dump()
- for intf in interfaces:
- self.vapi.nat66_add_del_interface(is_add=0, flags=intf.flags,
- sw_if_index=intf.sw_if_index)
-
- static_mappings = self.vapi.nat66_static_mapping_dump()
- for sm in static_mappings:
- self.vapi.nat66_add_del_static_mapping(
- local_ip_address=sm.local_ip_address,
- external_ip_address=sm.external_ip_address, vrf_id=sm.vrf_id,
- is_add=0)
-
- def tearDown(self):
- super(TestNAT66, self).tearDown()
- self.clear_nat66()
-
- def show_commands_at_teardown(self):
- self.logger.info(self.vapi.cli("show nat66 interfaces"))
- self.logger.info(self.vapi.cli("show nat66 static mappings"))
-
-
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff