aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_sixrd.py
blob: eca05452e3b5914aaaa713350acd9239d3cbebb7 (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
4
# -*- coding: utf-8 -*-

# daemon/pidfile.py
# Part of ‘python-daemon’, an implementation of PEP 3143.
#
# Copyright © 2008–2015 Ben Finney <ben+python@benfinney.id.au>
#
# This is free software: you may copy, modify, and/or distribute this work
# under the terms of the Apache License, version 2.0 as published by the
# Apache Software Foundation.
# No warranty expressed or implied. See the file ‘LICENSE.ASF-2’ for details.

""" Lockfile behaviour implemented via Unix PID files.
    """

from __future__ import (absolute_import, unicode_literals)

from lockfile.pidlockfile import PIDLockFile


class TimeoutPIDLockFile(PIDLockFile, object):
    """ Lockfile with default timeout, implemented as a Unix PID file.

        This uses the ``PIDLockFile`` implementation, with the
        following changes:

        * The `acquire_timeout` parameter to the initialiser will be
          used as the default `timeout` parameter for the `acquire`
          method.

        """

    def __init__(self, path, acquire_timeout=None, *args, **kwargs):
        """ Set up the parameters of a TimeoutPIDLockFile.

            :param path: Filesystem path to the PID file.
            :param acquire_timeout: Value to use by default for the
                `acquire` call.
            :return: ``None``.

            """
        self.acquire_timeout = acquire_timeout
        super(TimeoutPIDLockFile, self).__init__(path, *args, **kwargs)

    def acquire(self, timeout=None, *args, **kwargs):
        """ Acquire the lock.

            :param timeout: Specifies the timeout; see below for valid
                values.
            :return: ``None``.

            The `timeout` defaults to the value set during
            initialisation with the `acquire_timeout` parameter. It is
            passed to `PIDLockFile.acquire`; see that method for
            details.

            """
        if timeout is None:
            timeout = self.acquire_timeout
        super(TimeoutPIDLockFile, self).acquire(timeout, *args, **kwargs)


# Local variables:
# coding: utf-8
# mode: python
# End:
# vim: fileencoding=utf-8 filetype=python :
id='n385' href='#n385'>385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422
#!/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
from asfframework import VppTestRunner
from vpp_ip_route import VppIpRoute, VppRoutePath, VppIpTable

""" 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)