aboutsummaryrefslogtreecommitdiffstats
path: root/test/test_ipsec_api.py
blob: 521762b8181399305367ae94a854824c5657af97 (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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
import unittest

from framework import VppTestCase, VppTestRunner
from template_ipsec import TemplateIpsec, IPsecIPv4Params
from vpp_papi import VppEnum

from vpp_ipsec import VppIpsecSA


class IpsecApiTestCase(VppTestCase):
    """IPSec API tests"""

    vpp_worker_count = 2

    @classmethod
    def setUpClass(cls):
        super(IpsecApiTestCase, cls).setUpClass()

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

    def setUp(self):
        super(IpsecApiTestCase, self).setUp()
        self.create_pg_interfaces([0])
        self.pg0.config_ip4()
        self.pg0.admin_up()

        self.vpp_esp_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP
        self.vpp_ah_protocol = VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_AH
        self.ipv4_params = IPsecIPv4Params()

    def tearDown(self):
        self.pg0.unconfig_ip4()
        self.pg0.admin_down()
        super(IpsecApiTestCase, self).tearDown()

    def test_backend_dump(self):
        """backend dump"""
        d = self.vapi.ipsec_backend_dump()
        self.assert_equal(len(d), 2, "number of ipsec backends in dump")
        self.assert_equal(
            d[0].protocol, self.vpp_ah_protocol, "ipsec protocol in dump entry"
        )
        self.assert_equal(d[0].index, 0, "index in dump entry")
        self.assert_equal(d[0].active, 1, "active flag in dump entry")
        self.assert_equal(
            d[1].protocol, self.vpp_esp_protocol, "ipsec protocol in dump entry"
        )
        self.assert_equal(d[1].index, 0, "index in dump entry")
        self.assert_equal(d[1].active, 1, "active flag in dump entry")

    def test_select_valid_backend(self):
        """select valid backend"""
        self.vapi.ipsec_select_backend(self.vpp_ah_protocol, 0)
        self.vapi.ipsec_select_backend(self.vpp_esp_protocol, 0)

    def test_select_invalid_backend(self):
        """select invalid backend"""
        with self.vapi.assert_negative_api_retval():
            self.vapi.ipsec_select_backend(self.vpp_ah_protocol, 200)
        with self.vapi.assert_negative_api_retval():
            self.vapi.ipsec_select_backend(self.vpp_esp_protocol, 200)

    def test_select_backend_in_use(self):
        """attempt to change backend while sad configured"""
        params = self.ipv4_params
        addr_type = params.addr_type
        is_ipv6 = params.is_ipv6
        scapy_tun_sa_id = params.scapy_tun_sa_id
        scapy_tun_spi = params.scapy_tun_spi
        auth_algo_vpp_id = params.auth_algo_vpp_id
        auth_key = params.auth_key
        crypt_algo_vpp_id = params.crypt_algo_vpp_id
        crypt_key = params.crypt_key

        self.vapi.ipsec_sad_entry_add_del(
            is_add=1,
            entry={
                "sad_id": scapy_tun_sa_id,
                "spi": scapy_tun_spi,
                "integrity_algorithm": auth_algo_vpp_id,
                "integrity_key": {
                    "data": auth_key,
                    "length": len(auth_key),
                },
                "crypto_algorithm": crypt_algo_vpp_id,
                "crypto_key": {
                    "data": crypt_key,
                    "length": len(crypt_key),
                },
                "protocol": self.vpp_ah_protocol,
                "tunnel_src": self.pg0.local_addr[addr_type],
                "tunnel_dst": self.pg0.remote_addr[addr_type],
            },
        )
        with self.vapi.assert_negative_api_retval():
            self.vapi.ipsec_select_backend(protocol=self.vpp_ah_protocol, index=0)

        self.vapi.ipsec_sad_entry_add_del(
            is_add=0,
            entry={
                "sad_id": scapy_tun_sa_id,
                "spi": scapy_tun_spi,
                "integrity_algorithm": auth_algo_vpp_id,
                "integrity_key": {
                    "data": auth_key,
                    "length": len(auth_key),
                },
                "crypto_algorithm": crypt_algo_vpp_id,
                "crypto_key": {
                    "data": crypt_key,
                    "length": len(crypt_key),
                },
                "protocol": self.vpp_ah_protocol,
                "tunnel_src": self.pg0.local_addr[addr_type],
                "tunnel_dst": self.pg0.remote_addr[addr_type],
            },
        )
        self.vapi.ipsec_select_backend(protocol=self.vpp_ah_protocol, index=0)

    def __check_sa_binding(self, sa_id, thread_index):
        found_sa = False
        sa_dumps = self.vapi.ipsec_sa_v4_dump()
        for dump in sa_dumps:
            if dump.entry.sad_id == sa_id:
                self.assertEqual(dump.thread_index, thread_index)
                found_sa = True
                break

        if not found_sa:
            self.fail("SA not found in VPP")

    def test_sa_worker_bind(self):
        """Bind an SA to a worker"""
        sa = VppIpsecSA(
            self,
            self.ipv4_params.scapy_tun_sa_id,
            self.ipv4_params.scapy_tun_spi,
            self.ipv4_params.auth_algo_vpp_id,
            self.ipv4_params.auth_key,
            self.ipv4_params.crypt_algo_vpp_id,
            self.ipv4_params.crypt_key,
            VppEnum.vl_api_ipsec_proto_t.IPSEC_API_PROTO_ESP,
        )
        sa.add_vpp_config()

        self.__check_sa_binding(sa.id, 0xFFFF)

        self.vapi.ipsec_sad_bind(sa_id=sa.id, worker=1)

        self.__check_sa_binding(sa.id, 2)

        sa.remove_vpp_config()


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