"""IPsec utilities library."""
import os
+from random import choice
+from string import letters
from ipaddress import ip_network, ip_address
from enum import Enum, IntEnum
-from robot.api import logger
from resources.libraries.python.PapiExecutor import PapiExecutor
-from resources.libraries.python.PapiErrors import PapiError
from resources.libraries.python.topology import Topology
from resources.libraries.python.VatExecutor import VatExecutor
from resources.libraries.python.VatJsonUtil import VatJsonUtil
+def gen_key(length):
+ """Generate random string as a key.
+
+ :param length: Length of generated payload.
+ :type length: int
+ :returns: The generated payload.
+ :rtype: str
+ """
+ return ''.join(choice(letters) for _ in range(length)).encode('hex')
+
class PolicyAction(Enum):
"""Policy actions."""
BYPASS = 'bypass'
class CryptoAlg(Enum):
"""Encryption algorithms."""
AES_CBC_128 = ('aes-cbc-128', 'AES-CBC', 16)
- AES_CBC_192 = ('aes-cbc-192', 'AES-CBC', 24)
AES_CBC_256 = ('aes-cbc-256', 'AES-CBC', 32)
- AES_GCM_128 = ('aes-gcm-128', 'AES-GCM', 20)
+ AES_GCM_128 = ('aes-gcm-128', 'AES-GCM', 16)
+ AES_GCM_256 = ('aes-gcm-256', 'AES-GCM', 32)
def __init__(self, alg_name, scapy_name, key_len):
self.alg_name = alg_name
class IntegAlg(Enum):
"""Integrity algorithm."""
- SHA1_96 = ('sha1-96', 'HMAC-SHA1-96', 20)
SHA_256_128 = ('sha-256-128', 'SHA2-256-128', 32)
- SHA_384_192 = ('sha-384-192', 'SHA2-384-192', 48)
SHA_512_256 = ('sha-512-256', 'SHA2-512-256', 64)
- AES_GCM_128 = ('aes-gcm-128', 'AES-GCM', 20)
+ AES_GCM_128 = ('aes-gcm-128', 'AES-GCM', 16)
+ AES_GCM_256 = ('aes-gcm-256', 'AES-GCM', 32)
def __init__(self, alg_name, scapy_name, key_len):
self.alg_name = alg_name
"""
return CryptoAlg.AES_GCM_128
+ @staticmethod
+ def crypto_alg_aes_gcm_256():
+ """Return encryption algorithm aes-gcm-256.
+
+ :returns: CryptoAlg enum AES_GCM_128 object.
+ :rtype: CryptoAlg
+ """
+ return CryptoAlg.AES_GCM_256
+
@staticmethod
def get_crypto_alg_key_len(crypto_alg):
"""Return encryption algorithm key length.
"""
return IntegAlg.AES_GCM_128
+ @staticmethod
+ def integ_alg_aes_gcm_256():
+ """Return integrity algorithm AES-GCM-256.
+
+ :returns: IntegAlg enum AES_GCM_256 object.
+ :rtype: IntegAlg
+ """
+ return IntegAlg.AES_GCM_256
+
@staticmethod
def get_integ_alg_key_len(integ_alg):
"""Return integrity algorithm key length.
:type node: dict
:type protocol: IPsecProto
:type index: int
+ :raises RuntimeError: If failed to select IPsec backend or if no API
+ reply received.
"""
- # TODO: move composition of api data to separate method
- api_data = list()
- api = dict(api_name='ipsec_select_backend')
- api_args = dict(protocol=protocol)
- api_args['index'] = index
- api['api_args'] = api_args
- api_data.append(api)
-
- api_reply = None
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- except AssertionError:
- raise PapiError('Failed to select IPsec backend on host {host}'.
- format(host=node['host']))
- api_reply = papi_executor.get_papi_reply()
-
- if api_reply is not None:
- api_r = api_reply[0]['api_reply']['ipsec_select_backend_reply']
- if api_r['retval'] == 0:
- logger.trace('IPsec backend successfully selected on host '
- '{host}'.format(host=node['host']))
- else:
- raise PapiError('Failed to select IPsec backend on host {host}'.
- format(host=node['host']))
- else:
- raise PapiError('No reply received for ipsec_select_backend API '
- 'command on host {host}'.format(host=node['host']))
+
+ cmd = 'ipsec_select_backend'
+ cmd_reply = 'ipsec_select_backend_reply'
+ err_msg = 'Failed to select IPsec backend on host {host}'.format(
+ host=node['host'])
+ args = dict(protocol=protocol, index=index)
+ with PapiExecutor(node) as papi_exec:
+ papi_resp = papi_exec.add(cmd, **args).execute_should_pass(err_msg)
+ data = papi_resp.reply[0]['api_reply'][cmd_reply]
+ if data['retval'] != 0:
+ raise RuntimeError('Failed to select IPsec backend on host {host}'.
+ format(host=node['host']))
@staticmethod
def vpp_ipsec_backend_dump(node):
:param node: VPP node to dump IPsec backend on.
:type node: dict
"""
- # TODO: move composition of api data to separate method
- api_data = list()
- api = dict(api_name='ipsec_backend_dump')
- api_args = dict()
- api['api_args'] = api_args
- api_data.append(api)
-
- api_reply = None
- with PapiExecutor(node) as papi_executor:
- papi_executor.execute_papi(api_data)
- try:
- papi_executor.papi_should_have_passed()
- except AssertionError:
- raise PapiError('Failed to dump IPsec backends on host {host}'.
- format(host=node['host']))
- # After API change there is returned VPP internal enum object
- # representing VPP IPSEC protocol instead of integer representation
- # so JSON fails to decode it - we need to check if it is Python API
- # bug or we need to adapt vpp_papi_provider to correctly encode
- # such object into JSON
- # api_reply = papi_executor.get_papi_reply()
- api_reply = papi_executor.get_papi_stdout()
-
- if api_reply is not None:
- logger.trace('IPsec backend dump\n{dump}'.format(dump=api_reply))
- else:
- raise PapiError('No reply received for ipsec_select_backend API '
- 'command on host {host}'.format(host=node['host']))
+
+ err_msg = 'Failed to dump IPsec backends on host {host}'.format(
+ host=node['host'])
+ with PapiExecutor(node) as papi_exec:
+ papi_exec.add('ipsec_backend_dump').execute_should_pass(
+ err_msg, process_reply=False)
@staticmethod
def vpp_ipsec_add_sad_entry(node, sad_id, spi, crypto_alg, crypto_key,
:type tunnel_dst: str
"""
tmp_filename = '/tmp/ipsec_sad_{0}_add_del_entry.script'.format(sad_id)
- ckey = crypto_key.encode('hex')
- ikey = integ_key.encode('hex')
- tunnel = 'tunnel-src {0} tunnel-dst {1}'.format(tunnel_src, tunnel_dst)\
- if tunnel_src is not None and tunnel_dst is not None else ''
- integ = 'integ-alg {0} integ-key {1}'.format(integ_alg.alg_name, ikey)\
- if crypto_alg.alg_name != 'aes-gcm-128' else ''
+ addr_incr = 1 << (32 - 24)
with open(tmp_filename, 'w') as tmp_file:
for i in range(0, n_entries):
- buf_str = 'exec ipsec sa add {0} esp spi {1} ' \
- 'crypto-alg {2} crypto-key {3} {4} {5}\n'.format(
- sad_id+i, spi+i, crypto_alg.alg_name, ckey, integ,
- tunnel)
- tmp_file.write(buf_str)
+ integ = ''
+ if not crypto_alg.alg_name.startswith('aes-gcm-'):
+ integ = (
+ 'integ-alg {integ_alg} integ-key {integ_key}'.
+ format(
+ integ_alg=integ_alg.alg_name,
+ integ_key=integ_key))
+ tunnel = (
+ 'tunnel-src {laddr} tunnel-dst {raddr}'.
+ format(
+ laddr=ip_address(unicode(tunnel_src)) + i * addr_incr,
+ raddr=ip_address(unicode(tunnel_dst)) + i * addr_incr)
+ if tunnel_src and tunnel_dst is not None else '')
+ conf = (
+ 'exec ipsec sa add {sad_id} esp spi {spi} '
+ 'crypto-alg {crypto_alg} crypto-key {crypto_key} '
+ '{integ} {tunnel}\n'.
+ format(
+ sad_id=sad_id + i,
+ spi=spi + i,
+ crypto_alg=crypto_alg.alg_name,
+ crypto_key=crypto_key,
+ integ=integ,
+ tunnel=tunnel))
+ tmp_file.write(conf)
+ vat = VatExecutor()
+ vat.execute_script(tmp_filename, node, timeout=300, json_out=False,
+ copy_on_execute=True)
+ os.remove(tmp_filename)
+
+ @staticmethod
+ def vpp_ipsec_set_ip_route(node, n_tunnels, tunnel_src, traffic_addr,
+ tunnel_dst, interface, raddr_range):
+ """Set IP address and route on interface.
+
+ :param node: VPP node to add config on.
+ :param n_tunnels: Number of tunnels to create.
+ :param tunnel_src: Tunnel header source IPv4 or IPv6 address.
+ :param traffic_addr: Traffic destination IP address to route.
+ :param tunnel_dst: Tunnel header destination IPv4 or IPv6 address.
+ :param interface: Interface key on node 1.
+ :param raddr_range: Mask specifying range of Policy selector Remote IPv4
+ addresses. Valid values are from 1 to 32.
+ :type node: dict
+ :type n_tunnels: int
+ :type tunnel_src: str
+ :type traffic_addr: str
+ :type tunnel_dst: str
+ :type interface: str
+ :type raddr_range: int
+ """
+ tmp_filename = '/tmp/ipsec_set_ip.script'
+
+ addr_incr = 1 << (32 - raddr_range)
+
+ with open(tmp_filename, 'w') as tmp_file:
+ for i in range(0, n_tunnels):
+ conf = (
+ 'exec set interface ip address {interface} {laddr}/24\n'
+ 'exec ip route add {taddr}/32 via {raddr} {interface}\n'.
+ format(
+ interface=Topology.get_interface_name(node, interface),
+ laddr=ip_address(unicode(tunnel_src)) + i * addr_incr,
+ raddr=ip_address(unicode(tunnel_dst)) + i * addr_incr,
+ taddr=ip_address(unicode(traffic_addr)) + i))
+ tmp_file.write(conf)
vat = VatExecutor()
vat.execute_script(tmp_filename, node, timeout=300, json_out=False,
copy_on_execute=True)
@staticmethod
def vpp_ipsec_spd_add_entries(node, n_entries, spd_id, priority, inbound,
- sa_id, raddr_ip, raddr_range):
+ sa_id, raddr_ip):
"""Create multiple Security Policy Database entries on the VPP node.
:param node: VPP node to add SPD entries on.
entry. Remote IPv4 end address will be calculated depending on
raddr_range parameter. Each subsequent entry will have start address
next after IPv4 end address of previous entry.
- :param raddr_range: Mask specifying range of Policy selector Remote IPv4
- addresses. Valid values are from 1 to 32.
:type node: dict
:type n_entries: int
:type spd_id: int
:type inbound: bool
:type sa_id: int
:type raddr_ip: string
- :type raddr_range: int
"""
tmp_filename = '/tmp/ipsec_spd_{0}_add_del_entry.script'.format(sa_id)
- direction = 'inbound' if inbound else 'outbound'
- addr_incr = 1 << (32 - raddr_range)
- addr_ip = int(ip_address(unicode(raddr_ip)))
- start_str = 'exec ipsec policy add spd {0} priority {1} {2} ' \
- 'action protect sa'.format(spd_id, priority, direction)
with open(tmp_filename, 'w') as tmp_file:
for i in range(0, n_entries):
- r_ip_s = ip_address(addr_ip + addr_incr * i)
- r_ip_e = ip_address(addr_ip + addr_incr * (i+1) - 1)
- buf_str = '{0} {1} remote-ip-range {2} - {3}\n'.format(
- start_str, sa_id+i, r_ip_s, r_ip_e)
- tmp_file.write(buf_str)
+ raddr_s = ip_address(unicode(raddr_ip)) + i
+ raddr_e = ip_address(unicode(raddr_ip)) + (i + 1) - 1
+ tunnel = (
+ 'exec ipsec policy add spd {spd_id} priority {priority} '
+ '{direction} action protect sa {sa_id} '
+ 'remote-ip-range {raddr_s} - {raddr_e}\n'.
+ format(
+ spd_id=spd_id,
+ priority=priority,
+ direction='inbound' if inbound else 'outbound',
+ sa_id=sa_id+i,
+ raddr_s=raddr_s,
+ raddr_e=raddr_e))
+ tmp_file.write(tunnel)
vat = VatExecutor()
vat.execute_script(tmp_filename, node, timeout=300, json_out=False,
copy_on_execute=True)
os.remove(tmp_filename)
@staticmethod
- def vpp_ipsec_create_tunnel_interfaces(node1, node2, if1_ip_addr,
- if2_ip_addr, if1_key, if2_key,
- n_tunnels, crypto_alg, crypto_key,
- integ_alg, integ_key, raddr_ip1,
+ def vpp_ipsec_create_tunnel_interfaces(nodes, if1_ip_addr, if2_ip_addr,
+ if1_key, if2_key, n_tunnels,
+ crypto_alg, integ_alg, raddr_ip1,
raddr_ip2, raddr_range):
"""Create multiple IPsec tunnel interfaces between two VPP nodes.
- :param node1: VPP node 1 to create tunnel interfaces.
- :param node2: VPP node 2 to create tunnel interfaces.
+ :param nodes: VPP nodes to create tunnel interfaces.
:param if1_ip_addr: VPP node 1 interface IP4 address.
:param if2_ip_addr: VPP node 2 interface IP4 address.
:param if1_key: VPP node 1 interface key from topology file.
:param if2_key: VPP node 2 interface key from topology file.
:param n_tunnels: Number of tunnell interfaces to create.
:param crypto_alg: The encryption algorithm name.
- :param crypto_key: The encryption key string.
:param integ_alg: The integrity algorithm name.
- :param integ_key: The integrity key string.
:param raddr_ip1: Policy selector remote IPv4 start address for the
first tunnel in direction node1->node2.
:param raddr_ip2: Policy selector remote IPv4 start address for the
first tunnel in direction node2->node1.
:param raddr_range: Mask specifying range of Policy selector Remote IPv4
addresses. Valid values are from 1 to 32.
- :type node1: dict
- :type node2: dict
+ :type nodes: dict
:type if1_ip_addr: str
:type if2_ip_addr: str
:type if1_key: str
:type if2_key: str
:type n_tunnels: int
:type crypto_alg: CryptoAlg
- :type crypto_key: str
:type integ_alg: IntegAlg
- :type integ_key: str
:type raddr_ip1: string
:type raddr_ip2: string
:type raddr_range: int
"""
- spi_1 = 10000
- spi_2 = 20000
-
- raddr_ip1_i = int(ip_address(unicode(raddr_ip1)))
- raddr_ip2_i = int(ip_address(unicode(raddr_ip2)))
+ spi_1 = 100000
+ spi_2 = 200000
addr_incr = 1 << (32 - raddr_range)
tmp_fn1 = '/tmp/ipsec_create_tunnel_dut1.config'
tmp_fn2 = '/tmp/ipsec_create_tunnel_dut2.config'
- ckey = crypto_key.encode('hex')
- ikey = integ_key.encode('hex')
-
vat = VatExecutor()
+
with open(tmp_fn1, 'w') as tmp_f1, open(tmp_fn2, 'w') as tmp_f2:
for i in range(0, n_tunnels):
+ ckey = gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg))
+ ikey = gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg))
integ = ''
if not crypto_alg.alg_name.startswith('aes-gcm-'):
- integ = 'integ_alg {integ_alg} '\
- 'local_integ_key {local_integ_key} '\
- 'remote_integ_key {remote_integ_key} '\
- .format(integ_alg=integ_alg.alg_name,
- local_integ_key=ikey,
- remote_integ_key=ikey)
- dut1_tunnel = 'ipsec_tunnel_if_add_del '\
- 'local_spi {local_spi} '\
- 'remote_spi {remote_spi} '\
- 'crypto_alg {crypto_alg} '\
- 'local_crypto_key {local_crypto_key} '\
- 'remote_crypto_key {remote_crypto_key} '\
- '{integ} '\
- 'local_ip {local_ip} '\
- 'remote_ip {remote_ip}\n'\
- .format(local_spi=spi_1+i,
- remote_spi=spi_2+i,
- crypto_alg=crypto_alg.alg_name,
- local_crypto_key=ckey,
- remote_crypto_key=ckey,
- integ=integ,
- local_ip=if1_ip_addr,
- remote_ip=if2_ip_addr)
- dut2_tunnel = 'ipsec_tunnel_if_add_del '\
- 'local_spi {local_spi} '\
- 'remote_spi {remote_spi} '\
- 'crypto_alg {crypto_alg} '\
- 'local_crypto_key {local_crypto_key} '\
- 'remote_crypto_key {remote_crypto_key} '\
- '{integ} '\
- 'local_ip {local_ip} '\
- 'remote_ip {remote_ip}\n'\
- .format(local_spi=spi_2+i,
- remote_spi=spi_1+i,
- crypto_alg=crypto_alg.alg_name,
- local_crypto_key=ckey,
- remote_crypto_key=ckey,
- integ=integ,
- local_ip=if2_ip_addr,
- remote_ip=if1_ip_addr)
- tmp_f1.write(dut1_tunnel)
- tmp_f2.write(dut2_tunnel)
- vat.execute_script(tmp_fn1, node1, timeout=300, json_out=False,
+ integ = (
+ 'integ_alg {integ_alg} '
+ 'local_integ_key {local_integ_key} '
+ 'remote_integ_key {remote_integ_key} '
+ .format(
+ integ_alg=integ_alg.alg_name,
+ local_integ_key=ikey,
+ remote_integ_key=ikey))
+ tmp_f1.write(
+ 'exec set interface ip address {uifc} {laddr}/24\n'
+ 'ipsec_tunnel_if_add_del '
+ 'local_spi {local_spi} '
+ 'remote_spi {remote_spi} '
+ 'crypto_alg {crypto_alg} '
+ 'local_crypto_key {local_crypto_key} '
+ 'remote_crypto_key {remote_crypto_key} '
+ '{integ} '
+ 'local_ip {laddr} '
+ 'remote_ip {raddr}\n'
+ .format(
+ local_spi=spi_1 + i,
+ remote_spi=spi_2 + i,
+ crypto_alg=crypto_alg.alg_name,
+ local_crypto_key=ckey,
+ remote_crypto_key=ckey,
+ integ=integ,
+ laddr=ip_address(unicode(if1_ip_addr)) + i * addr_incr,
+ raddr=ip_address(unicode(if2_ip_addr)) + i * addr_incr,
+ uifc=Topology.get_interface_name(nodes['DUT1'],
+ if1_key)))
+ tmp_f2.write(
+ 'exec set interface ip address {uifc} {laddr}/24\n'
+ 'ipsec_tunnel_if_add_del '
+ 'local_spi {local_spi} '
+ 'remote_spi {remote_spi} '
+ 'crypto_alg {crypto_alg} '
+ 'local_crypto_key {local_crypto_key} '
+ 'remote_crypto_key {remote_crypto_key} '
+ '{integ} '
+ 'local_ip {laddr} '
+ 'remote_ip {raddr}\n'
+ .format(
+ local_spi=spi_2 + i,
+ remote_spi=spi_1 + i,
+ crypto_alg=crypto_alg.alg_name,
+ local_crypto_key=ckey,
+ remote_crypto_key=ckey,
+ integ=integ,
+ laddr=ip_address(unicode(if2_ip_addr)) + i * addr_incr,
+ raddr=ip_address(unicode(if1_ip_addr)) + i * addr_incr,
+ uifc=Topology.get_interface_name(nodes['DUT2'],
+ if2_key)))
+ vat.execute_script(tmp_fn1, nodes['DUT1'], timeout=300, json_out=False,
copy_on_execute=True)
- vat.execute_script(tmp_fn2, node2, timeout=300, json_out=False,
+ vat.execute_script(tmp_fn2, nodes['DUT2'], timeout=300, json_out=False,
copy_on_execute=True)
os.remove(tmp_fn1)
os.remove(tmp_fn2)
with open(tmp_fn1, 'w') as tmp_f1, open(tmp_fn2, 'w') as tmp_f2:
for i in range(0, n_tunnels):
- raddr_ip1 = ip_address(raddr_ip1_i + addr_incr*i)
- raddr_ip2 = ip_address(raddr_ip2_i + addr_incr*i)
- dut1_if = Topology.get_interface_name(node1, if1_key)
- dut1 = 'exec ip route add {raddr}/{mask} via {addr} ipsec{i}\n'\
- 'exec set interface unnumbered ipsec{i} use {uifc}\n'\
- 'exec set interface state ipsec{i} up\n'\
- .format(raddr=raddr_ip2, mask=raddr_range,
- addr=if2_ip_addr, i=i, uifc=dut1_if)
- dut2_if = Topology.get_interface_name(node2, if2_key)
- dut2 = 'exec ip route add {raddr}/{mask} via {addr} ipsec{i}\n'\
- 'exec set interface unnumbered ipsec{i} use {uifc}\n'\
- 'exec set interface state ipsec{i} up\n'\
- .format(raddr=raddr_ip1, mask=raddr_range,
- addr=if1_ip_addr, i=i, uifc=dut2_if)
- tmp_f1.write(dut1)
- tmp_f2.write(dut2)
-
- vat.execute_script(tmp_fn1, node1, timeout=300, json_out=False,
+ tmp_f1.write(
+ 'exec set interface unnumbered ipsec{i} use {uifc}\n'
+ 'exec set interface state ipsec{i} up\n'
+ 'exec ip route add {taddr}/32 via {raddr} ipsec{i}\n'
+ .format(
+ taddr=ip_address(unicode(raddr_ip2)) + i,
+ raddr=ip_address(unicode(if2_ip_addr)) + i * addr_incr,
+ i=i,
+ uifc=Topology.get_interface_name(nodes['DUT1'],
+ if1_key)))
+ tmp_f2.write(
+ 'exec set interface unnumbered ipsec{i} use {uifc}\n'
+ 'exec set interface state ipsec{i} up\n'
+ 'exec ip route add {taddr}/32 via {raddr} ipsec{i}\n'
+ .format(
+ taddr=ip_address(unicode(raddr_ip1)) + i,
+ raddr=ip_address(unicode(if1_ip_addr)) + i * addr_incr,
+ i=i,
+ uifc=Topology.get_interface_name(nodes['DUT2'],
+ if2_key)))
+ vat.execute_script(tmp_fn1, nodes['DUT1'], timeout=300, json_out=False,
copy_on_execute=True)
- vat.execute_script(tmp_fn2, node2, timeout=300, json_out=False,
+ vat.execute_script(tmp_fn2, nodes['DUT2'], timeout=300, json_out=False,
copy_on_execute=True)
os.remove(tmp_fn1)
os.remove(tmp_fn2)
@staticmethod
- def vpp_ipsec_add_multiple_tunnels(node1, node2, interface1, interface2,
- n_tunnels, crypto_alg, crypto_key,
- integ_alg, integ_key, tunnel_ip1,
- tunnel_ip2, raddr_ip1, raddr_ip2,
- raddr_range):
+ def vpp_ipsec_add_multiple_tunnels(nodes, interface1, interface2,
+ n_tunnels, crypto_alg, integ_alg,
+ tunnel_ip1, tunnel_ip2, raddr_ip1,
+ raddr_ip2, raddr_range):
"""Create multiple IPsec tunnels between two VPP nodes.
- :param node1: VPP node 1 to create tunnels.
- :param node2: VPP node 2 to create tunnels.
+ :param nodes: VPP nodes to create tunnels.
:param interface1: Interface name or sw_if_index on node 1.
:param interface2: Interface name or sw_if_index on node 2.
:param n_tunnels: Number of tunnels to create.
:param crypto_alg: The encryption algorithm name.
- :param crypto_key: The encryption key string.
:param integ_alg: The integrity algorithm name.
- :param integ_key: The integrity key string.
:param tunnel_ip1: Tunnel node1 IPv4 address.
:param tunnel_ip2: Tunnel node2 IPv4 address.
:param raddr_ip1: Policy selector remote IPv4 start address for the
first tunnel in direction node2->node1.
:param raddr_range: Mask specifying range of Policy selector Remote IPv4
addresses. Valid values are from 1 to 32.
- :type node1: dict
- :type node2: dict
+ :type nodes: dict
:type interface1: str or int
:type interface2: str or int
:type n_tunnels: int
:type crypto_alg: CryptoAlg
- :type crypto_key: str
:type integ_alg: str
- :type integ_key: str
:type tunnel_ip1: str
:type tunnel_ip2: str
:type raddr_ip1: string
spd_id = 1
p_hi = 100
p_lo = 10
- sa_id_1 = 10000
- sa_id_2 = 20000
- spi_1 = 30000
- spi_2 = 40000
- proto = 50
-
- IPsecUtil.vpp_ipsec_add_spd(node1, spd_id)
- IPsecUtil.vpp_ipsec_spd_add_if(node1, spd_id, interface1)
- IPsecUtil.vpp_ipsec_policy_add(node1, spd_id, p_hi, PolicyAction.BYPASS,
- inbound=False, proto=proto)
- IPsecUtil.vpp_ipsec_policy_add(node1, spd_id, p_hi, PolicyAction.BYPASS,
- inbound=True, proto=proto)
-
- IPsecUtil.vpp_ipsec_add_spd(node2, spd_id)
- IPsecUtil.vpp_ipsec_spd_add_if(node2, spd_id, interface2)
- IPsecUtil.vpp_ipsec_policy_add(node2, spd_id, p_hi, PolicyAction.BYPASS,
- inbound=False, proto=proto)
- IPsecUtil.vpp_ipsec_policy_add(node2, spd_id, p_hi, PolicyAction.BYPASS,
- inbound=True, proto=proto)
-
- IPsecUtil.vpp_ipsec_add_sad_entries(node1, n_tunnels, sa_id_1, spi_1,
- crypto_alg, crypto_key, integ_alg,
- integ_key, tunnel_ip1, tunnel_ip2)
-
- IPsecUtil.vpp_ipsec_spd_add_entries(node1, n_tunnels, spd_id, p_lo,
- False, sa_id_1, raddr_ip2,
- raddr_range)
-
- IPsecUtil.vpp_ipsec_add_sad_entries(node2, n_tunnels, sa_id_1, spi_1,
- crypto_alg, crypto_key, integ_alg,
- integ_key, tunnel_ip1, tunnel_ip2)
-
- IPsecUtil.vpp_ipsec_spd_add_entries(node2, n_tunnels, spd_id, p_lo,
- True, sa_id_1, raddr_ip2,
- raddr_range)
-
- IPsecUtil.vpp_ipsec_add_sad_entries(node2, n_tunnels, sa_id_2, spi_2,
- crypto_alg, crypto_key, integ_alg,
- integ_key, tunnel_ip2, tunnel_ip1)
-
- IPsecUtil.vpp_ipsec_spd_add_entries(node2, n_tunnels, spd_id, p_lo,
- False, sa_id_2, raddr_ip1,
- raddr_range)
-
- IPsecUtil.vpp_ipsec_add_sad_entries(node1, n_tunnels, sa_id_2, spi_2,
- crypto_alg, crypto_key, integ_alg,
- integ_key, tunnel_ip2, tunnel_ip1)
-
- IPsecUtil.vpp_ipsec_spd_add_entries(node1, n_tunnels, spd_id, p_lo,
- True, sa_id_2, raddr_ip1,
- raddr_range)
+ sa_id_1 = 100000
+ sa_id_2 = 200000
+ spi_1 = 300000
+ spi_2 = 400000
+
+ crypto_key = gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg))
+ integ_key = gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg))
+
+ IPsecUtil.vpp_ipsec_set_ip_route(
+ nodes['DUT1'], n_tunnels, tunnel_ip1, raddr_ip2, tunnel_ip2,
+ interface1, raddr_range)
+ IPsecUtil.vpp_ipsec_set_ip_route(
+ nodes['DUT2'], n_tunnels, tunnel_ip2, raddr_ip1, tunnel_ip1,
+ interface2, raddr_range)
+
+ IPsecUtil.vpp_ipsec_add_spd(
+ nodes['DUT1'], spd_id)
+ IPsecUtil.vpp_ipsec_spd_add_if(
+ nodes['DUT1'], spd_id, interface1)
+ IPsecUtil.vpp_ipsec_policy_add(
+ nodes['DUT1'], spd_id, p_hi, PolicyAction.BYPASS, inbound=False,
+ proto=50)
+ IPsecUtil.vpp_ipsec_policy_add(
+ nodes['DUT1'], spd_id, p_hi, PolicyAction.BYPASS, inbound=True,
+ proto=50)
+
+ IPsecUtil.vpp_ipsec_add_spd(
+ nodes['DUT2'], spd_id)
+ IPsecUtil.vpp_ipsec_spd_add_if(
+ nodes['DUT2'], spd_id, interface2)
+ IPsecUtil.vpp_ipsec_policy_add(
+ nodes['DUT2'], spd_id, p_hi, PolicyAction.BYPASS, inbound=False,
+ proto=50)
+ IPsecUtil.vpp_ipsec_policy_add(
+ nodes['DUT2'], spd_id, p_hi, PolicyAction.BYPASS, inbound=True,
+ proto=50)
+
+ IPsecUtil.vpp_ipsec_add_sad_entries(
+ nodes['DUT1'], n_tunnels, sa_id_1, spi_1, crypto_alg, crypto_key,
+ integ_alg, integ_key, tunnel_ip1, tunnel_ip2)
+
+ IPsecUtil.vpp_ipsec_spd_add_entries(
+ nodes['DUT1'], n_tunnels, spd_id, p_lo, False, sa_id_1, raddr_ip2)
+
+ IPsecUtil.vpp_ipsec_add_sad_entries(
+ nodes['DUT2'], n_tunnels, sa_id_1, spi_1, crypto_alg, crypto_key,
+ integ_alg, integ_key, tunnel_ip1, tunnel_ip2)
+
+ IPsecUtil.vpp_ipsec_spd_add_entries(
+ nodes['DUT2'], n_tunnels, spd_id, p_lo, True, sa_id_1, raddr_ip2)
+
+ IPsecUtil.vpp_ipsec_add_sad_entries(
+ nodes['DUT2'], n_tunnels, sa_id_2, spi_2, crypto_alg, crypto_key,
+ integ_alg, integ_key, tunnel_ip2, tunnel_ip1)
+
+ IPsecUtil.vpp_ipsec_spd_add_entries(
+ nodes['DUT2'], n_tunnels, spd_id, p_lo, False, sa_id_2, raddr_ip1)
+
+ IPsecUtil.vpp_ipsec_add_sad_entries(
+ nodes['DUT1'], n_tunnels, sa_id_2, spi_2, crypto_alg, crypto_key,
+ integ_alg, integ_key, tunnel_ip2, tunnel_ip1)
+
+ IPsecUtil.vpp_ipsec_spd_add_entries(
+ nodes['DUT1'], n_tunnels, spd_id, p_lo, True, sa_id_2, raddr_ip1)
@staticmethod
def vpp_ipsec_show(node):
Code Review / csit.git / blobdiff