aboutsummaryrefslogtreecommitdiffstats
path: root/resources/libraries/bash/entry/bootstrap_verify_perf.sh
blob: 2329beba700d0386ebb3e19498270a8276c26e97 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
# Copyright (c) 2019 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

set -exuo pipefail

# Assumptions:
# + There is a directory holding CSIT code to use (this script is there).
# + At least one of the following is true:
# ++ JOB_NAME environment variable is set,
# ++ or this entry script has access to arguments.
# Consequences (and specific assumptions) are multiple,
# examine tree of functions for current description.

# FIXME: Define API contract (as opposed to just help) for bootstrap.

# "set -eu" handles failures from the following two lines.
BASH_ENTRY_DIR="$(dirname $(readlink -e "${BASH_SOURCE[0]}"))"
BASH_FUNCTION_DIR="$(readlink -e "${BASH_ENTRY_DIR}/../function")"
source "${BASH_FUNCTION_DIR}/common.sh" || {
    echo "Source failed." >&2
    exit 1
}
source "${BASH_FUNCTION_DIR}/gather.sh" || die "Source failed."
source "${BASH_FUNCTION_DIR}/ansible.sh" || die "Source failed."
common_dirs || die
check_prerequisites || die
get_test_code "${1-}" || die
get_test_tag_string || die
select_topology || die
select_arch_os || die
gather_build || die
check_download_dir || die
activate_virtualenv || die
generate_tests || die
archive_tests || die
reserve_and_cleanup_testbed || die
ansible_hosts "calibration" || die
select_tags || die
compose_pybot_arguments || die
run_pybot || die
untrap_and_unreserve_testbed || die
copy_archives || die
die_on_pybot_error || die
ground-color: #fff0f0 } /* Literal.String.Heredoc */ .highlight .si { color: #3333bb; background-color: #fff0f0 } /* Literal.String.Interpol */ .highlight .sx { color: #22bb22; background-color: #f0fff0 } /* Literal.String.Other */ .highlight .sr { color: #008800; background-color: #fff0ff } /* Literal.String.Regex */ .highlight .s1 { color: #dd2200; background-color: #fff0f0 } /* Literal.String.Single */ .highlight .ss { color: #aa6600; background-color: #fff0f0 } /* Literal.String.Symbol */ .highlight .bp { color: #003388 } /* Name.Builtin.Pseudo */ .highlight .fm { color: #0066bb; font-weight: bold } /* Name.Function.Magic */ .highlight .vc { color: #336699 } /* Name.Variable.Class */ .highlight .vg { color: #dd7700 } /* Name.Variable.Global */ .highlight .vi { color: #3333bb } /* Name.Variable.Instance */ .highlight .vm { color: #336699 } /* Name.Variable.Magic */ .highlight .il { color: #0000DD; font-weight: bold } /* Literal.Number.Integer.Long */ }
#!/usr/bin/env python3
""" 6RD RFC5969 functional tests """

import unittest
from scapy.layers.inet import IP, UDP, Ether
from scapy.layers.inet6 import IPv6
from scapy.packet import Raw
from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable, FibPathProto
from socket import AF_INET, AF_INET6, inet_pton

""" Test6rd is a subclass of  VPPTestCase classes.

6RD tests.

"""


class Test6RD(VppTestCase):
    """6RD Test Case"""

    @classmethod
    def setUpClass(cls):
        super(Test6RD, cls).setUpClass()
        cls.create_pg_interfaces(range(4))
        cls.interfaces = list(cls.pg_interfaces)

    @classmethod
    def tearDownClass(cls):
        super(Test6RD, cls).tearDownClass()

    def setUp(self):
        super(Test6RD, self).setUp()
        t4 = VppIpTable(self, 10)
        t6 = VppIpTable(self, 20, True)

        t4.add_vpp_config()
        t6.add_vpp_config()

        for n in range(4):
            i = self.pg_interfaces[n]
            i.admin_up()
            if n > 1:
                i.set_table_ip4(10)
                i.set_table_ip6(20)
            i.config_ip4()
            i.config_ip6()
            i.disable_ipv6_ra()
            i.resolve_arp()
            i.resolve_ndp()

    def tearDown(self):
        for i in self.pg_interfaces:
            i.unconfig_ip4()
            i.unconfig_ip6()
            i.set_table_ip4(0)
            i.set_table_ip6(0)
            i.admin_down()
        super(Test6RD, self).tearDown()

    def validate_6in4(self, rx, expected):
        if IP not in rx:
            self.fail()
        if IPv6 not in rx:
            self.fail()

        self.assertEqual(rx[IP].src, expected[IP].src)
        self.assertEqual(rx[IP].dst, expected[IP].dst)
        self.assertEqual(rx[IP].proto, expected[IP].proto)
        self.assertEqual(rx[IPv6].src, expected[IPv6].src)
        self.assertEqual(rx[IPv6].dst, expected[IPv6].dst)

    def validate_4in6(self, rx, expected):
        if IPv6 not in rx:
            self.fail()
        if IP in rx:
            self.fail()

        self.assertEqual(rx[IPv6].src, expected[IPv6].src)
        self.assertEqual(rx[IPv6].dst, expected[IPv6].dst)
        self.assertEqual(rx[IPv6].nh, expected[IPv6].nh)

    def payload(self, len):
        return "x" * len

    def test_6rd_ip6_to_ip4(self):
        """ip6 -> ip4 (encap) 6rd test"""
        p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
        p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1", nh="UDP")

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )

        self.tunnel_index = rv.sw_if_index

        self.vapi.cli("show ip6 fib")
        p_payload = UDP(sport=1234, dport=1234)
        p = p_ether / p_ip6 / p_payload

        p_reply = (
            IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, proto="ipv6") / p_ip6
        )

        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_6in4(p, p_reply)

        # MTU tests (default is 1480)
        plen = 1481 - 40 - 8
        p_ip6 = IPv6(src="1::1", dst="2002:AC10:0202::1")
        p_payload = UDP(sport=1234, dport=1234) / Raw(self.payload(plen))
        p = p_ether / p_ip6 / p_payload

        p_reply = (
            IP(src=self.pg0.local_ip4, dst=self.pg1.remote_ip4, proto="ipv6") / p_ip6
        )

        rx = self.send_and_assert_no_replies(self.pg0, p * 10)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_ip6_to_ip4_vrf(self):
        """ip6 -> ip4 (encap) 6rd VRF test"""
        p_ether = Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
        p_ip6 = IPv6(src="1::1", dst="2002:AC10:0402::1", nh="UDP")

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=20,
            ip4_table_id=10,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg2.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index

        self.vapi.cli("show ip6 fib")
        p_payload = UDP(sport=1234, dport=1234)
        p = p_ether / p_ip6 / p_payload

        p_reply = (
            IP(src=self.pg2.local_ip4, dst=self.pg3.remote_ip4, proto="ipv6") / p_ip6
        )

        rx = self.send_and_expect(self.pg2, p * 10, self.pg3)
        for p in rx:
            self.validate_6in4(p, p_reply)

        # MTU tests (default is 1480)
        plen = 1481 - 40 - 8
        p_ip6 = IPv6(src="1::1", dst="2002:AC10:0402::1")
        p_payload = UDP(sport=1234, dport=1234) / Raw(self.payload(plen))
        p = p_ether / p_ip6 / p_payload

        p_reply = (
            IP(src=self.pg2.local_ip4, dst=self.pg3.remote_ip4, proto="ipv6") / p_ip6
        )

        rx = self.send_and_assert_no_replies(self.pg0, p * 10)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_ip4_to_ip6(self):
        """ip4 -> ip6 (decap) 6rd test"""

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index
        rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index)
        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index

        p_ip6 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )

        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
            / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
            / p_ip6
        )

        p_reply = p_ip6

        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_reply)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_ip4_to_ip6_vrf(self):
        """ip4 -> ip6 (decap) 6rd VRF test"""

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=20,
            ip4_table_id=10,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg2.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index
        rv = self.vapi.ipip_6rd_del_tunnel(rv.sw_if_index)
        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=20,
            ip4_table_id=10,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg2.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index
        self.vapi.sw_interface_set_table(self.tunnel_index, 1, 20)

        p_ip6 = IPv6(src="2002:AC10:0402::1", dst=self.pg3.remote_ip6) / UDP(
            sport=1234, dport=1234
        )

        p = (
            Ether(src=self.pg2.remote_mac, dst=self.pg2.local_mac)
            / IP(src=self.pg3.remote_ip4, dst=self.pg2.local_ip4)
            / p_ip6
        )

        p_reply = p_ip6

        rx = self.send_and_expect(self.pg2, p * 10, self.pg3)
        for p in rx:
            self.validate_4in6(p, p_reply)
        self.vapi.sw_interface_set_table(self.tunnel_index, 1, 0)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_ip4_to_ip6_multiple(self):
        """ip4 -> ip6 (decap) 6rd test"""

        self.tunnel_index = []
        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )
        self.tunnel_index.append(rv.sw_if_index)
        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2003::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg1.local_ip4,
            security_check=True,
        )

        self.tunnel_index.append(rv.sw_if_index)

        self.vapi.cli("show ip6 fib")
        p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
        p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
        p_ip6_1 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )
        p_ip6_2 = IPv6(src="2003:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )

        p = p_ether / p_ip4 / p_ip6_1
        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_ip6_1)

        p = p_ether / p_ip4 / p_ip6_2
        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_ip6_2)
        for i in self.tunnel_index:
            self.vapi.ipip_6rd_del_tunnel(i)

    def test_6rd_ip4_to_ip6_suffix(self):
        """ip4 -> ip6 (decap) 6rd test"""

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="172.0.0.0/8",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index

        self.vapi.cli("show ip6 fib")
        p_ether = Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
        p_ip4 = IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
        p_ip6 = IPv6(src="2002:1002:0200::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )

        p = p_ether / p_ip4 / p_ip6
        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_ip6)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_ip4_to_ip6_sec_check(self):
        """ip4 -> ip6 (decap) security check 6rd test"""

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=True,
        )
        self.tunnel_index = rv.sw_if_index

        self.vapi.cli("show ip6 fib")
        p_ip6 = IPv6(src="2002:AC10:0202::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )
        p_ip6_fail = IPv6(src="2002:DEAD:0202::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )

        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
            / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
            / p_ip6
        )

        p_reply = p_ip6

        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_reply)

        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
            / IP(src=self.pg1.remote_ip4, dst=self.pg0.local_ip4)
            / p_ip6_fail
        )
        rx = self.send_and_assert_no_replies(self.pg0, p * 10)
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)

    def test_6rd_bgp_tunnel(self):
        """6rd BGP tunnel"""

        rv = self.vapi.ipip_6rd_add_tunnel(
            ip6_table_id=0,
            ip4_table_id=0,
            ip6_prefix="2002::/16",
            ip4_prefix="0.0.0.0/0",
            ip4_src=self.pg0.local_ip4,
            security_check=False,
        )

        self.tunnel_index = rv.sw_if_index

        default_route = VppIpRoute(
            self, "DEAD::", 16, [VppRoutePath("2002:0808:0808::", self.tunnel_index)]
        )
        default_route.add_vpp_config()

        ip4_route = VppIpRoute(
            self, "8.0.0.0", 8, [VppRoutePath(self.pg1.remote_ip4, 0xFFFFFFFF)]
        )
        ip4_route.add_vpp_config()

        # Via recursive route 6 -> 4
        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
            / IPv6(src="1::1", dst="DEAD:BEEF::1")
            / UDP(sport=1234, dport=1234)
        )

        p_reply = IP(src=self.pg0.local_ip4, dst="8.8.8.8", proto="ipv6") / IPv6(
            src="1::1", dst="DEAD:BEEF::1", nh="UDP"
        )

        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_6in4(p, p_reply)

        # Via recursive route 4 -> 6 (Security check must be disabled)
        p_ip6 = IPv6(src="DEAD:BEEF::1", dst=self.pg1.remote_ip6) / UDP(
            sport=1234, dport=1234
        )
        p = (
            Ether(src=self.pg0.remote_mac, dst=self.pg0.local_mac)
            / IP(src="8.8.8.8", dst=self.pg0.local_ip4)
            / p_ip6
        )

        p_reply = p_ip6

        rx = self.send_and_expect(self.pg0, p * 10, self.pg1)
        for p in rx:
            self.validate_4in6(p, p_reply)
        ip4_route.remove_vpp_config()
        default_route.remove_vpp_config()
        self.vapi.ipip_6rd_del_tunnel(self.tunnel_index)


if __name__ == "__main__":
    unittest.main(testRunner=VppTestRunner)