import struct
from framework import VppTestCase, VppTestRunner, running_extended_tests
+from vpp_ip_route import VppIpRoute, VppRoutePath, DpoProto
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 ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from time import sleep
from util import ip4_range
+from util import mactobinary
class MethodHolder(VppTestCase):
cls.ipfix_src_port = 4739
cls.ipfix_domain_id = 1
- cls.create_pg_interfaces(range(9))
+ cls.create_pg_interfaces(range(10))
cls.interfaces = list(cls.pg_interfaces[0:4])
for i in cls.interfaces:
cls.pg0.configure_ipv4_neighbors()
cls.overlapping_interfaces = list(list(cls.pg_interfaces[4:7]))
+ cls.vapi.ip_table_add_del(10, is_add=1)
+ cls.vapi.ip_table_add_del(20, is_add=1)
cls.pg4._local_ip4 = "172.16.255.1"
cls.pg4._local_ip4n = socket.inet_pton(socket.AF_INET, i.local_ip4)
cls.pg7.admin_up()
cls.pg8.admin_up()
+ cls.pg9.generate_remote_hosts(2)
+ cls.pg9.config_ip4()
+ ip_addr_n = socket.inet_pton(socket.AF_INET, "10.0.0.1")
+ cls.vapi.sw_interface_add_del_address(cls.pg9.sw_if_index,
+ ip_addr_n,
+ 24)
+ cls.pg9.admin_up()
+ cls.pg9.resolve_arp()
+ cls.pg9._remote_hosts[1]._ip4 = cls.pg9._remote_hosts[0]._ip4
+ cls.pg4._remote_ip4 = cls.pg9._remote_hosts[0]._ip4 = "10.0.0.2"
+ cls.pg9.resolve_arp()
+
except Exception:
super(TestNAT44, cls).tearDownClass()
raise
interfaces = self.vapi.nat44_interface_dump()
for intf in interfaces:
+ if intf.is_inside > 1:
+ self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
+ 0,
+ is_add=0)
self.vapi.nat44_interface_add_del_feature(intf.sw_if_index,
intf.is_inside,
is_add=0)
lb_sm.external_port,
lb_sm.protocol,
lb_sm.vrf_id,
- is_add=0)
+ is_add=0,
+ local_num=0,
+ locals=[])
adresses = self.vapi.nat44_address_dump()
for addr in adresses:
self.pg0.unconfig_ip4()
self.pg1.unconfig_ip4()
+ self.vapi.ip_table_add_del(vrf_id1, is_add=1)
+ self.vapi.ip_table_add_del(vrf_id2, is_add=1)
self.pg0.set_table_ip4(vrf_id1)
self.pg1.set_table_ip4(vrf_id2)
self.pg0.config_ip4()
capture = self.pg2.get_capture(len(pkts))
self.verify_capture_out(capture, nat_ip2)
+ self.pg0.unconfig_ip4()
+ self.pg1.unconfig_ip4()
+ self.pg0.set_table_ip4(0)
+ self.pg1.set_table_ip4(0)
+ self.vapi.ip_table_add_del(vrf_id1, is_add=0)
+ self.vapi.ip_table_add_del(vrf_id2, is_add=0)
+
def test_vrf_feature_independent(self):
""" NAT44 tenant VRF independent address pool mode """
""" NAT44 interfaces without configured IP address """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
+ mactobinary(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,
+ mactobinary(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
""" NAT44 interfaces without configured IP address - 1:1 NAT """
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
+ mactobinary(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,
+ mactobinary(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
self.icmp_id_out = 30608
self.vapi.ip_neighbor_add_del(self.pg7.sw_if_index,
- self.pg7.remote_mac,
+ mactobinary(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,
+ mactobinary(self.pg8.remote_mac),
self.pg8.remote_ip4n,
is_static=1)
""" NAT44 interface output feature (in2out postrouting) """
self.nat44_add_address(self.nat_addr)
self.vapi.nat44_interface_add_del_output_feature(self.pg0.sw_if_index)
- self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index,
+ self.vapi.nat44_interface_add_del_output_feature(self.pg1.sw_if_index)
+ self.vapi.nat44_interface_add_del_output_feature(self.pg3.sw_if_index,
is_inside=0)
# in2out
- pkts = self.create_stream_in(self.pg0, self.pg1)
+ 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.pg1.get_capture(len(pkts))
+ capture = self.pg3.get_capture(len(pkts))
self.verify_capture_out(capture)
# out2in
- pkts = self.create_stream_out(self.pg1)
- self.pg1.add_stream(pkts)
+ 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)
+ # from non-NAT interface to NAT inside interface
+ pkts = self.create_stream_in(self.pg2, self.pg0)
+ self.pg2.add_stream(pkts)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(len(pkts))
+ self.verify_capture_no_translation(capture, self.pg2, self.pg0)
+
def test_output_feature_vrf_aware(self):
""" NAT44 interface output feature VRF aware (in2out postrouting) """
nat_ip_vrf10 = "10.0.0.10"
self.logger.error(ppp("Unexpected or invalid packet:"), p)
raise
+ def test_one_armed_nat44(self):
+ """ One armed NAT44 """
+ remote_host = self.pg9.remote_hosts[0]
+ local_host = self.pg9.remote_hosts[1]
+ external_port = 0
+
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg9.sw_if_index,
+ is_inside=0)
+
+ # in2out
+ p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
+ IP(src=local_host.ip4, dst=remote_host.ip4) /
+ TCP(sport=12345, dport=80))
+ self.pg9.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg9.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, self.nat_addr)
+ self.assertEqual(ip.dst, remote_host.ip4)
+ self.assertNotEqual(tcp.sport, 12345)
+ external_port = tcp.sport
+ self.assertEqual(tcp.dport, 80)
+ self.check_tcp_checksum(p)
+ self.check_ip_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ # out2in
+ p = (Ether(src=self.pg9.remote_mac, dst=self.pg9.local_mac) /
+ IP(src=remote_host.ip4, dst=self.nat_addr) /
+ TCP(sport=80, dport=external_port))
+ self.pg9.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg9.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, remote_host.ip4)
+ self.assertEqual(ip.dst, local_host.ip4)
+ self.assertEqual(tcp.sport, 80)
+ self.assertEqual(tcp.dport, 12345)
+ self.check_tcp_checksum(p)
+ self.check_ip_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ def test_del_session(self):
+ """ Delete NAT44 session """
+ self.nat44_add_address(self.nat_addr)
+ self.vapi.nat44_interface_add_del_feature(self.pg0.sw_if_index)
+ self.vapi.nat44_interface_add_del_feature(self.pg1.sw_if_index,
+ is_inside=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))
+
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+ nsessions = len(sessions)
+
+ self.vapi.nat44_del_session(sessions[0].inside_ip_address,
+ sessions[0].inside_port,
+ sessions[0].protocol)
+ self.vapi.nat44_del_session(sessions[1].outside_ip_address,
+ sessions[1].outside_port,
+ sessions[1].protocol,
+ is_in=0)
+
+ sessions = self.vapi.nat44_user_session_dump(self.pg0.remote_ip4n, 0)
+ self.assertEqual(nsessions - len(sessions), 2)
+
def tearDown(self):
super(TestNAT44, self).tearDown()
if not self.vpp_dead:
cls.vrf1_nat_addr_n = socket.inet_pton(socket.AF_INET,
cls.vrf1_nat_addr)
- cls.create_pg_interfaces(range(3))
+ cls.create_pg_interfaces(range(4))
cls.ip6_interfaces = list(cls.pg_interfaces[0:1])
cls.ip6_interfaces.append(cls.pg_interfaces[2])
cls.ip4_interfaces = list(cls.pg_interfaces[1:2])
+ cls.vapi.ip_table_add_del(cls.vrf1_id, is_add=1, is_ipv6=1)
+
cls.pg_interfaces[2].set_table_ip6(cls.vrf1_id)
cls.pg0.generate_remote_hosts(2)
i.config_ip4()
i.resolve_arp()
+ cls.pg3.admin_up()
+ cls.pg3.config_ip4()
+ cls.pg3.resolve_arp()
+ cls.pg3.config_ip6()
+ cls.pg3.configure_ipv6_neighbors()
+
except Exception:
super(TestNAT64, cls).tearDownClass()
raise
self.logger.error(ppp("Unexpected or invalid packet:", packet))
raise
+ def test_one_armed_nat64(self):
+ """ One armed NAT64 """
+ external_port = 0
+ remote_host_ip6 = self.compose_ip6(self.pg3.remote_ip4,
+ '64:ff9b::',
+ 96)
+
+ self.vapi.nat64_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ self.vapi.nat64_add_del_interface(self.pg3.sw_if_index)
+ self.vapi.nat64_add_del_interface(self.pg3.sw_if_index, is_inside=0)
+
+ # in2out
+ p = (Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) /
+ IPv6(src=self.pg3.remote_ip6, dst=remote_host_ip6) /
+ TCP(sport=12345, dport=80))
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IP]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, self.nat_addr)
+ self.assertEqual(ip.dst, self.pg3.remote_ip4)
+ self.assertNotEqual(tcp.sport, 12345)
+ external_port = tcp.sport
+ self.assertEqual(tcp.dport, 80)
+ self.check_tcp_checksum(p)
+ self.check_ip_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
+ # out2in
+ p = (Ether(src=self.pg3.remote_mac, dst=self.pg3.local_mac) /
+ IP(src=self.pg3.remote_ip4, dst=self.nat_addr) /
+ TCP(sport=80, dport=external_port))
+ self.pg3.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg3.get_capture(1)
+ p = capture[0]
+ try:
+ ip = p[IPv6]
+ tcp = p[TCP]
+ self.assertEqual(ip.src, remote_host_ip6)
+ self.assertEqual(ip.dst, self.pg3.remote_ip6)
+ self.assertEqual(tcp.sport, 80)
+ self.assertEqual(tcp.dport, 12345)
+ self.check_tcp_checksum(p)
+ except:
+ self.logger.error(ppp("Unexpected or invalid packet:", p))
+ raise
+
def nat64_get_ses_num(self):
"""
Return number of active NAT64 sessions.
interfaces = self.vapi.nat64_interface_dump()
for intf in interfaces:
+ if intf.is_inside > 1:
+ self.vapi.nat64_add_del_interface(intf.sw_if_index,
+ 0,
+ is_add=0)
self.vapi.nat64_add_del_interface(intf.sw_if_index,
intf.is_inside,
is_add=0)
self.logger.info(self.vapi.cli("show nat64 session table all"))
self.clear_nat64()
+
+class TestDSlite(MethodHolder):
+ """ DS-Lite Test Cases """
+
+ @classmethod
+ def setUpClass(cls):
+ super(TestDSlite, cls).setUpClass()
+
+ try:
+ 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.pg0.admin_up()
+ cls.pg0.config_ip4()
+ cls.pg0.resolve_arp()
+ cls.pg1.admin_up()
+ cls.pg1.config_ip6()
+ cls.pg1.generate_remote_hosts(2)
+ cls.pg1.configure_ipv6_neighbors()
+
+ except Exception:
+ super(TestDSlite, cls).tearDownClass()
+ raise
+
+ def test_dslite(self):
+ """ Test DS-Lite """
+ self.vapi.dslite_add_del_pool_addr_range(self.nat_addr_n,
+ self.nat_addr_n)
+ aftr_ip4 = '192.0.0.1'
+ aftr_ip4_n = socket.inet_pton(socket.AF_INET, aftr_ip4)
+ aftr_ip6 = '2001:db8:85a3::8a2e:370:1'
+ aftr_ip6_n = socket.inet_pton(socket.AF_INET6, aftr_ip6)
+ self.vapi.dslite_set_aftr_addr(aftr_ip6_n, aftr_ip4_n)
+
+ # UDP
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[0].ip6) /
+ IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+ UDP(sport=20000, dport=10000))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ capture = capture[0]
+ self.assertFalse(capture.haslayer(IPv6))
+ self.assertEqual(capture[IP].src, self.nat_addr)
+ self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+ self.assertNotEqual(capture[UDP].sport, 20000)
+ self.assertEqual(capture[UDP].dport, 10000)
+ self.check_ip_checksum(capture)
+ out_port = capture[UDP].sport
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+ UDP(sport=10000, dport=out_port))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ capture = capture[0]
+ self.assertEqual(capture[IPv6].src, aftr_ip6)
+ self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
+ self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+ self.assertEqual(capture[IP].dst, '192.168.1.1')
+ self.assertEqual(capture[UDP].sport, 10000)
+ self.assertEqual(capture[UDP].dport, 20000)
+ self.check_ip_checksum(capture)
+
+ # TCP
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
+ IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+ TCP(sport=20001, dport=10001))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ capture = capture[0]
+ self.assertFalse(capture.haslayer(IPv6))
+ self.assertEqual(capture[IP].src, self.nat_addr)
+ self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+ self.assertNotEqual(capture[TCP].sport, 20001)
+ self.assertEqual(capture[TCP].dport, 10001)
+ self.check_ip_checksum(capture)
+ self.check_tcp_checksum(capture)
+ out_port = capture[TCP].sport
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+ TCP(sport=10001, dport=out_port))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ capture = capture[0]
+ self.assertEqual(capture[IPv6].src, aftr_ip6)
+ self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
+ self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+ self.assertEqual(capture[IP].dst, '192.168.1.1')
+ self.assertEqual(capture[TCP].sport, 10001)
+ self.assertEqual(capture[TCP].dport, 20001)
+ self.check_ip_checksum(capture)
+ self.check_tcp_checksum(capture)
+
+ # ICMP
+ p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
+ IPv6(dst=aftr_ip6, src=self.pg1.remote_hosts[1].ip6) /
+ IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
+ ICMP(id=4000, type='echo-request'))
+ self.pg1.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg0.get_capture(1)
+ capture = capture[0]
+ self.assertFalse(capture.haslayer(IPv6))
+ self.assertEqual(capture[IP].src, self.nat_addr)
+ self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
+ self.assertNotEqual(capture[ICMP].id, 4000)
+ self.check_ip_checksum(capture)
+ self.check_icmp_checksum(capture)
+ out_id = capture[ICMP].id
+
+ p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
+ IP(dst=self.nat_addr, src=self.pg0.remote_ip4) /
+ ICMP(id=out_id, type='echo-reply'))
+ self.pg0.add_stream(p)
+ self.pg_enable_capture(self.pg_interfaces)
+ self.pg_start()
+ capture = self.pg1.get_capture(1)
+ capture = capture[0]
+ self.assertEqual(capture[IPv6].src, aftr_ip6)
+ self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[1].ip6)
+ self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
+ self.assertEqual(capture[IP].dst, '192.168.1.1')
+ self.assertEqual(capture[ICMP].id, 4000)
+ self.check_ip_checksum(capture)
+ self.check_icmp_checksum(capture)
+
+ def tearDown(self):
+ super(TestDSlite, self).tearDown()
+ if not self.vpp_dead:
+ self.logger.info(self.vapi.cli("show dslite pool"))
+ self.logger.info(
+ self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
+ self.logger.info(self.vapi.cli("show dslite sessions"))
+
if __name__ == '__main__':
unittest.main(testRunner=VppTestRunner)
Code Review / vpp.git / blobdiff