ligang/vpp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
vpp/test/test_dslite.py
Dmitry Valter 34fa0ce8f7 tests: skip more excluded plugin tests 
Check and skip VPP_EXCLUDED_PLUGINS tests for most of plugins.

Type: improvement
Signed-off-by: Dmitry Valter <d-valter@yandex-team.com>
Change-Id: I23fd3666729251c639aa8da72a676058e3f5bb4e
2024-07-12 15:43:24 +00:00

348 lines
13 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
import socket
import unittest
from asfframework import tag_fixme_vpp_workers
from framework import VppTestCase
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet6 import (
IPv6,
ICMPv6EchoRequest,
ICMPv6EchoReply,
)
from scapy.layers.l2 import Ether
from scapy.data import IP_PROTOS
from scapy.packet import Raw
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogSeverity
from vpp_ip_route import VppIpRoute, VppRoutePath
from config import config
@tag_fixme_vpp_workers
@unittest.skipIf("nat" in config.excluded_plugins, "Exclude NAT plugin tests")
class TestDSlite(VppTestCase):
"""DS-Lite Test Cases"""
@classmethod
def setUpClass(cls):
super(TestDSlite, cls).setUpClass()
try:
cls.nat_addr = "10.0.0.3"
cls.create_pg_interfaces(range(3))
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()
cls.pg2.admin_up()
cls.pg2.config_ip4()
cls.pg2.resolve_arp()
except Exception:
super(TestDSlite, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestDSlite, cls).tearDownClass()
def verify_syslog_apmadd(self, data, isaddr, isport, xsaddr, xsport, sv6enc, proto):
message = data.decode("utf-8")
try:
message = SyslogMessage.parse(message)
except ParseError as e:
self.logger.error(e)
else:
self.assertEqual(message.severity, SyslogSeverity.info)
self.assertEqual(message.appname, "NAT")
self.assertEqual(message.msgid, "APMADD")
sd_params = message.sd.get("napmap")
self.assertTrue(sd_params is not None)
self.assertEqual(sd_params.get("IATYP"), "IPv4")
self.assertEqual(sd_params.get("ISADDR"), isaddr)
self.assertEqual(sd_params.get("ISPORT"), "%d" % isport)
self.assertEqual(sd_params.get("XATYP"), "IPv4")
self.assertEqual(sd_params.get("XSADDR"), xsaddr)
self.assertEqual(sd_params.get("XSPORT"), "%d" % xsport)
self.assertEqual(sd_params.get("PROTO"), "%d" % proto)
self.assertTrue(sd_params.get("SSUBIX") is not None)
self.assertEqual(sd_params.get("SV6ENC"), sv6enc)
def test_dslite(self):
"""Test DS-Lite"""
self.vapi.dslite_add_del_pool_addr_range(
start_addr=self.nat_addr, end_addr=self.nat_addr, is_add=1
)
aftr_ip4 = "192.0.0.1"
aftr_ip6 = "2001:db8:85a3::8a2e:370:1"
self.vapi.dslite_set_aftr_addr(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
self.vapi.syslog_set_sender(self.pg2.local_ip4, self.pg2.remote_ip4)
# 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.assert_packet_checksums_valid(capture)
out_port = capture[UDP].sport
capture = self.pg2.get_capture(1)
self.verify_syslog_apmadd(
capture[0][Raw].load,
"192.168.1.1",
20000,
self.nat_addr,
out_port,
self.pg1.remote_hosts[0].ip6,
IP_PROTOS.udp,
)
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.assert_packet_checksums_valid(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.assert_packet_checksums_valid(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.assert_packet_checksums_valid(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.assert_packet_checksums_valid(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.assert_packet_checksums_valid(capture)
# ping DS-Lite AFTR tunnel endpoint address
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_hosts[1].ip6, dst=aftr_ip6)
/ ICMPv6EchoRequest()
)
self.pg1.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.assertTrue(capture.haslayer(ICMPv6EchoReply))
b4s = self.statistics.get_counter("/dslite/total-b4s")
self.assertEqual(b4s[0][0], 2)
sessions = self.statistics.get_counter("/dslite/total-sessions")
self.assertEqual(sessions[0][0], 3)
def tearDown(self):
super(TestDSlite, self).tearDown()
def show_commands_at_teardown(self):
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"))
@unittest.skipIf("nat" in config.excluded_plugins, "Exclude NAT plugin tests")
class TestDSliteCE(VppTestCase):
"""DS-Lite CE Test Cases"""
@classmethod
def setUpConstants(cls):
super(TestDSliteCE, cls).setUpConstants()
cls.vpp_cmdline.extend(["dslite", "{", "ce", "}"])
@classmethod
def setUpClass(cls):
super(TestDSliteCE, cls).setUpClass()
try:
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(1)
cls.pg1.configure_ipv6_neighbors()
except Exception:
super(TestDSliteCE, cls).tearDownClass()
raise
@classmethod
def tearDownClass(cls):
super(TestDSliteCE, cls).tearDownClass()
def test_dslite_ce(self):
"""Test DS-Lite CE"""
b4_ip4 = "192.0.0.2"
b4_ip6 = "2001:db8:62aa::375e:f4c1:1"
self.vapi.dslite_set_b4_addr(ip4_addr=b4_ip4, ip6_addr=b4_ip6)
aftr_ip4 = "192.0.0.1"
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(ip4_addr=aftr_ip4, ip6_addr=aftr_ip6)
r1 = VppIpRoute(
self,
aftr_ip6,
128,
[VppRoutePath(self.pg1.remote_ip6, self.pg1.sw_if_index)],
)
r1.add_vpp_config()
# UDP encapsulation
p = (
Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
/ IP(dst=self.pg1.remote_ip4, src=self.pg0.remote_ip4)
/ UDP(sport=10000, dport=20000)
)
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, b4_ip6)
self.assertEqual(capture[IPv6].dst, aftr_ip6)
self.assertEqual(capture[IP].src, self.pg0.remote_ip4)
self.assertEqual(capture[IP].dst, self.pg1.remote_ip4)
self.assertEqual(capture[UDP].sport, 10000)
self.assertEqual(capture[UDP].dport, 20000)
self.assert_packet_checksums_valid(capture)
# UDP decapsulation
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(dst=b4_ip6, src=aftr_ip6)
/ IP(dst=self.pg0.remote_ip4, src=self.pg1.remote_ip4)
/ 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.pg1.remote_ip4)
self.assertEqual(capture[IP].dst, self.pg0.remote_ip4)
self.assertEqual(capture[UDP].sport, 20000)
self.assertEqual(capture[UDP].dport, 10000)
self.assert_packet_checksums_valid(capture)
# ping DS-Lite B4 tunnel endpoint address
p = (
Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
/ IPv6(src=self.pg1.remote_hosts[0].ip6, dst=b4_ip6)
/ ICMPv6EchoRequest()
)
self.pg1.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, b4_ip6)
self.assertEqual(capture[IPv6].dst, self.pg1.remote_hosts[0].ip6)
self.assertTrue(capture.haslayer(ICMPv6EchoReply))
def tearDown(self):
super(TestDSliteCE, self).tearDown()
def show_commands_at_teardown(self):
self.logger.info(self.vapi.cli("show dslite aftr-tunnel-endpoint-address"))
self.logger.info(self.vapi.cli("show dslite b4-tunnel-endpoint-address"))
Reference in New Issue View Git Blame Copy Permalink