add classify session action set-sr-policy-index - vpp

diff options

author	Gabriel Ganne <gabriel.ganne@enea.com>	2017-10-02 11:41:24 +0200
committer	Damjan Marion <dmarion.lists@gmail.com>	2017-11-10 20:25:10 +0000
commit	8527f12b525181b47cc7c0c9e943a743b1a3b19e (patch)
tree	a4a407a2d1b25fc42f80ff06b952271a72f60025 /src/vnet/cop
parent	f616d10d04d5c444e20e617841a54cfb2c58d07d (diff)

add classify session action set-sr-policy-index

This allows to use the classifier to steer source routing packets instead of using the "sr steer" command. This way we can steer on anything instead of only the dst ip address. test: * add add_node_next function to the VppPapiProvider class. * add simple test scenario using the classifier to steer packets with dest ip addr == a7::/8 to the source routing insert node. * use new interface indexes (3,4) instead of (0,1) to prevent a cleanup conflict with the other tests which attach a specific fib to the interface. The test creates interfaces sepsrated from the other tests to prevent a conflict in the cleaning of the ip6 fib index 1 which causes vpp not to be able to find a default route on this table. Change-Id: Ibacb30fab3ce53f0dfe848ca6a8cdf0d111d8336 Signed-off-by: Gabriel Ganne <gabriel.ganne@enea.com>

Diffstat (limited to 'src/vnet/cop')

0 files changed, 0 insertions, 0 deletions

import unittest from framework import VppTestCase from asfframework import VppTestRunner from template_ipsec import 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_v5_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)


context:
space:
mode: