test/test_dslite.py

#!/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"))