-#!/usr/bin/env python
+#!/usr/bin/env python3
import socket
import unittest
from vpp_papi import VppEnum
from vpp_ip_route import VppIpRoute, VppRoutePath, FibPathType
from vpp_neighbor import VppNeighbor
-from vpp_ip import VppIpAddress, VppIpPrefix
from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
PacketListField
cls.pg1.configure_ipv4_neighbors()
cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
- cls.vapi.ip_table_add_del(is_add=1, table_id=10)
- cls.vapi.ip_table_add_del(is_add=1, table_id=20)
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 10})
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 20})
- cls.pg4._local_ip4 = VppIpPrefix("172.16.255.1",
- cls.pg4.local_ip4_prefix.len)
+ cls.pg4._local_ip4 = "172.16.255.1"
cls.pg4._remote_hosts[0]._ip4 = "172.16.255.2"
cls.pg4.set_table_ip4(10)
- cls.pg5._local_ip4 = VppIpPrefix("172.17.255.3",
- cls.pg5.local_ip4_prefix.len)
+ cls.pg5._local_ip4 = "172.17.255.3"
cls.pg5._remote_hosts[0]._ip4 = "172.17.255.4"
cls.pg5.set_table_ip4(10)
- cls.pg6._local_ip4 = VppIpPrefix("172.16.255.1",
- cls.pg6.local_ip4_prefix.len)
+ cls.pg6._local_ip4 = "172.16.255.1"
cls.pg6._remote_hosts[0]._ip4 = "172.16.255.2"
cls.pg6.set_table_ip4(20)
for i in cls.overlapping_interfaces:
cls.pg9.config_ip4()
cls.vapi.sw_interface_add_del_address(
sw_if_index=cls.pg9.sw_if_index,
- prefix=VppIpPrefix("10.0.0.1", 24).encode())
+ prefix="10.0.0.1/24")
cls.pg9.admin_up()
cls.pg9.resolve_arp()
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.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)
-
- # get maximum number of translations per user
- nat44_config = self.vapi.nat_show_config()
-
- # send more than maximum number of translations per user packets
- pkts_num = nat44_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)
-
- users = self.vapi.nat44_user_dump()
- for user in users:
- if user.ip_address == self.pg0.remote_ip4:
- self.assertEqual(user.nsessions,
- nat44_config.max_translations_per_user)
- self.assertEqual(user.nstaticsessions, 0)
-
- tcp_port = 22
- self.nat44_add_static_mapping(self.pg0.remote_ip4, self.nat_addr,
- tcp_port, tcp_port,
- proto=IP_PROTOS.tcp)
- 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=tcp_port))
- self.pg0.add_stream(p)
- self.pg_enable_capture(self.pg_interfaces)
- self.pg_start()
- self.pg1.get_capture(1)
- users = self.vapi.nat44_user_dump()
- for user in users:
- if user.ip_address == self.pg0.remote_ip4:
- self.assertEqual(user.nsessions,
- nat44_config.max_translations_per_user - 1)
- self.assertEqual(user.nstaticsessions, 1)
-
def test_interface_addr(self):
""" Acquire NAT44 addresses from interface """
self.vapi.nat44_add_del_interface_addr(
self.pg0.unconfig_ip4()
self.pg1.unconfig_ip4()
- self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id1)
- self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id2)
+ self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
+ self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
self.pg0.set_table_ip4(vrf_id1)
self.pg1.set_table_ip4(vrf_id2)
self.pg0.config_ip4()
self.pg1.config_ip4()
self.pg0.resolve_arp()
self.pg1.resolve_arp()
- self.vapi.ip_table_add_del(is_add=0, table_id=vrf_id1)
- self.vapi.ip_table_add_del(is_add=0, table_id=vrf_id2)
+ self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id1})
+ self.vapi.ip_table_add_del(is_add=0, table={'table_id': vrf_id2})
def test_vrf_feature_independent(self):
""" NAT44 tenant VRF independent address pool mode """
self.pg1.unconfig_ip4()
self.pg2.unconfig_ip4()
- self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id1)
- self.vapi.ip_table_add_del(is_add=1, table_id=vrf_id2)
+ self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id1})
+ self.vapi.ip_table_add_del(is_add=1, table={'table_id': vrf_id2})
self.pg1.set_table_ip4(vrf_id1)
self.pg2.set_table_ip4(vrf_id2)
self.pg1.config_ip4()
self.logger.info(self.vapi.cli("show nat ha"))
+class TestNAT44EndpointDependent2(MethodHolder):
+ """ Endpoint-Dependent session test cases """
+
+ icmp_timeout = 2
+
+ @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), "}"])
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestNAT44EndpointDependent2, cls).setUpClass()
+ try:
+ translation_buckets = 1
+ cls.max_translations = 10 * translation_buckets
+
+ cls.create_pg_interfaces(range(2))
+ cls.interfaces = list(cls.pg_interfaces[0:2])
+
+ 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()
+
+ cls.pg1.generate_remote_hosts(1)
+ cls.pg1.configure_ipv4_neighbors()
+
+ except Exception:
+ super(TestNAT44EndpointDependent2, cls).tearDownClass()
+ raise
+
+ def create_icmp_stream(self, in_if, out_if, count):
+ """
+ Create ICMP packet stream for inside network
+
+ :param in_if: Inside interface
+ :param out_if: Outside interface
+ :param count: Number of packets
+ """
+
+ 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'))
+ 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_cleanup(self):
+ """ NAT44 session cleanup test """
+
+ 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)
+
+ nat_config = self.vapi.nat_show_config()
+ self.assertEqual(1, nat_config.endpoint_dependent)
+
+ pkts = self.create_icmp_stream(self.pg0, self.pg1,
+ self.max_translations + 2)
+ sz = len(pkts)
+
+ # positive test
+ self.send_pkts(pkts[0:self.max_translations])
+
+ # false positive test
+ self.send_pkts(pkts[self.max_translations:sz - 1], 0)
+
+ sleep(self.icmp_timeout)
+
+ # positive test
+ self.send_pkts(pkts[self.max_translations + 1:sz])
+
+
class TestNAT44EndpointDependent(MethodHolder):
""" Endpoint-Dependent mapping and filtering test cases """
cls.pg4.config_ip4()
cls.vapi.sw_interface_add_del_address(
sw_if_index=cls.pg4.sw_if_index,
- prefix=VppIpPrefix("10.0.0.1", 24).encode())
+ prefix="10.0.0.1/24")
cls.pg4.admin_up()
cls.pg4.resolve_arp()
cls.pg4.resolve_arp()
zero_ip4 = socket.inet_pton(socket.AF_INET, "0.0.0.0")
- cls.vapi.ip_table_add_del(is_add=1, table_id=1)
+ cls.vapi.ip_table_add_del(is_add=1, table={'table_id': 1})
- cls.pg5._local_ip4 = VppIpPrefix("10.1.1.1",
- cls.pg5.local_ip4_prefix.len)
+ cls.pg5._local_ip4 = "10.1.1.1"
cls.pg5._remote_hosts[0]._ip4 = "10.1.1.2"
cls.pg5.set_table_ip4(1)
cls.pg5.config_ip4()
register=False)
r1.add_vpp_config()
- cls.pg6._local_ip4 = VppIpPrefix("10.1.2.1",
- cls.pg6.local_ip4_prefix.len)
+ cls.pg6._local_ip4 = "10.1.2.1"
cls.pg6._remote_hosts[0]._ip4 = "10.1.2.2"
cls.pg6.set_table_ip4(1)
cls.pg6.config_ip4()
is_add=1)
try:
- self.vapi.ip_table_add_del(is_add=1, table_id=new_vrf_id)
+ self.vapi.ip_table_add_del(is_add=1,
+ table={'table_id': new_vrf_id})
self.pg7.unconfig_ip4()
self.pg7.set_table_ip4(new_vrf_id)
self.pg8.config_ip4()
self.pg8.resolve_arp()
- self.vapi.ip_table_add_del(is_add=0, table_id=new_vrf_id)
+ self.vapi.ip_table_add_del(is_add=0,
+ table={'table_id': new_vrf_id})
def test_forwarding(self):
""" NAT44 forwarding test """
cls.ip6_interfaces.append(cls.pg_interfaces[2])
cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
- cls.vapi.ip_table_add_del(is_ipv6=1, is_add=1,
- table_id=cls.vrf1_id)
+ cls.vapi.ip_table_add_del(is_add=1,
+ table={'table_id': cls.vrf1_id,
+ 'is_ip6': 1})
cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
Code Review / vpp.git / blobdiff