:type interface: str
:type raddr_range: int
"""
- laddr = ip_address(tunnel_src)
- raddr = ip_address(tunnel_dst)
- taddr = ip_address(traffic_addr)
- addr_incr = 1 << (128 - raddr_range) if laddr.version == 6 \
+ tunnel_src = ip_address(tunnel_src)
+ tunnel_dst = ip_address(tunnel_dst)
+ traffic_addr = ip_address(traffic_addr)
+ addr_incr = 1 << (128 - raddr_range) if tunnel_src.version == 6 \
else 1 << (32 - raddr_range)
if int(n_tunnels) > 10:
if_name = Topology.get_interface_name(node, interface)
for i in range(n_tunnels):
conf = f"exec set interface ip address {if_name} " \
- f"{laddr + i * addr_incr}/{raddr_range}\n" \
- f"exec ip route add {taddr + i}/" \
- f"{128 if taddr.version == 6 else 32} " \
- f"via {raddr + i * addr_incr} {if_name}\n"
+ f"{tunnel_src + i * addr_incr}/{raddr_range}\n" \
+ f"exec ip route add {traffic_addr + i}/" \
+ f"{128 if traffic_addr.version == 6 else 32} " \
+ f"via {tunnel_dst + i * addr_incr} {if_name}\n"
tmp_file.write(conf)
- vat = VatExecutor()
- vat.execute_script(
+ VatExecutor().execute_script(
tmp_filename, node, timeout=300, json_out=False,
copy_on_execute=True
)
with PapiSocketExecutor(node) as papi_exec:
for i in range(n_tunnels):
args1[u"prefix"] = IPUtil.create_prefix_object(
- laddr + i * addr_incr, raddr_range
+ tunnel_src + i * addr_incr, raddr_range
)
args2[u"route"] = IPUtil.compose_vpp_route_structure(
- node, taddr + i,
- prefix_len=128 if taddr.version == 6 else 32,
- interface=interface, gateway=raddr + i * addr_incr
+ node, traffic_addr + i,
+ prefix_len=128 if traffic_addr.version == 6 else 32,
+ interface=interface, gateway=tunnel_dst + i * addr_incr
)
history = bool(not 1 < i < n_tunnels - 2)
papi_exec.add(cmd1, history=history, **args1).\
priority=int(priority),
is_outbound=not inbound,
sa_id=int(sa_id) if sa_id else 0,
- policy=IPsecUtil.policy_action_protect().policy_int_repr,
+ policy=getattr(PolicyAction.PROTECT, u"policy_int_repr"),
protocol=0,
remote_address_start=IPAddress.create_ip_address_object(raddr_ip),
remote_address_stop=IPAddress.create_ip_address_object(raddr_ip),
papi_exec.get_replies(err_msg)
@staticmethod
- 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,
- existing_tunnels=0):
- """Create multiple IPsec tunnel interfaces between two VPP nodes.
+ def _ipsec_create_tunnel_interfaces_dut1_vat(
+ nodes, tun_ips, if1_key, if2_key, n_tunnels, crypto_alg, integ_alg,
+ raddr_ip2, addr_incr, spi_d, existing_tunnels=0):
+ """Create multiple IPsec tunnel interfaces on DUT1 node using VAT.
:param nodes: VPP nodes to create tunnel interfaces.
- :param if1_ip_addr: VPP node 1 interface IPv4/IPv6 address.
- :param if2_ip_addr: VPP node 2 interface IPv4/IPv6 address.
+ :param tun_ips: Dictionary with VPP node 1 ipsec tunnel interface
+ IPv4/IPv6 address (ip1) and VPP node 2 ipsec tunnel interface
+ IPv4/IPv6 address (ip2).
:param if1_key: VPP node 1 interface key from topology file.
- :param if2_key: VPP node 2 interface key from topology file.
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
:param n_tunnels: Number of tunnel interfaces to be there at the end.
:param crypto_alg: The encryption algorithm name.
:param integ_alg: The integrity algorithm name.
- :param raddr_ip1: Policy selector remote IPv4/IPv6 start address for the
- first tunnel in direction node1->node2.
:param raddr_ip2: Policy selector remote IPv4/IPv6 start address for the
first tunnel in direction node2->node1.
- :param raddr_range: Mask specifying range of Policy selector Remote
- IPv4/IPv6 addresses. Valid values are from 1 to 32 in case of IPv4
- and to 128 in case of IPv6.
+ :param spi_d: Dictionary with SPIs for VPP node 1 and VPP node 2.
+ :param addr_incr: IP / IPv6 address incremental step.
:param existing_tunnels: Number of tunnel interfaces before creation.
Useful mainly for reconf tests. Default 0.
:type nodes: dict
- :type if1_ip_addr: str
- :type if2_ip_addr: str
+ :type tun_ips: dict
:type if1_key: str
:type if2_key: str
:type n_tunnels: int
:type crypto_alg: CryptoAlg
:type integ_alg: IntegAlg
- :type raddr_ip1: string
- :type raddr_ip2: string
- :type raddr_range: int
+ :type raddr_ip2: IPv4Address or IPv6Address
+ :type addr_incr: int
+ :type spi_d: dict
:type existing_tunnels: int
"""
- n_tunnels = int(n_tunnels)
- existing_tunnels = int(existing_tunnels)
- spi_1 = 100000
- spi_2 = 200000
- if1_ip = ip_address(if1_ip_addr)
- if2_ip = ip_address(if2_ip_addr)
- raddr_ip1 = ip_address(raddr_ip1)
- raddr_ip2 = ip_address(raddr_ip2)
- addr_incr = 1 << (128 - raddr_range) if if1_ip.version == 6 \
- else 1 << (32 - raddr_range)
+ tmp_fn1 = u"/tmp/ipsec_create_tunnel_dut1.config"
+ if1_n = Topology.get_interface_name(nodes[u"DUT1"], if1_key)
- if n_tunnels - existing_tunnels > 10:
- tmp_fn1 = u"/tmp/ipsec_create_tunnel_dut1.config"
- tmp_fn2 = u"/tmp/ipsec_create_tunnel_dut2.config"
- if1_n = Topology.get_interface_name(nodes[u"DUT1"], if1_key)
- if2_n = Topology.get_interface_name(nodes[u"DUT2"], if2_key)
- mask = 96 if if2_ip.version == 6 else 24
- mask2 = 128 if if2_ip.version == 6 else 32
- vat = VatExecutor()
- with open(tmp_fn1, 'w') as tmp_f1, open(tmp_fn2, 'w') as tmp_f2:
- rmac = Topology.get_interface_mac(nodes[u"DUT2"], if2_key)
- if not existing_tunnels:
- tmp_f1.write(
- f"exec create loopback interface\n"
- f"exec set interface state loop0 up\n"
- f"exec set interface ip address "
- f"{if1_n} {if2_ip - 1}/{mask}\n"
- f"exec set ip neighbor {if1_n} {if2_ip}/{mask2} {rmac}"
- f" static\n"
- )
- tmp_f2.write(
- f"exec set interface ip address {if2_n}"
- f" {if2_ip}/{mask}\n"
- )
- for i in range(existing_tunnels, n_tunnels):
- ckey = gen_key(
- IPsecUtil.get_crypto_alg_key_len(crypto_alg)
- ).hex()
- if integ_alg:
- ikey = gen_key(
- IPsecUtil.get_integ_alg_key_len(integ_alg)
- ).hex()
- integ = f"integ_alg {integ_alg.alg_name} " \
- f"local_integ_key {ikey} remote_integ_key {ikey} "
- else:
- integ = u""
- tmp_f1.write(
- f"exec set interface ip address loop0 "
- f"{if1_ip + i * addr_incr}/32\n"
- f"ipsec_tunnel_if_add_del "
- f"local_spi {spi_1 + i} remote_spi {spi_2 + i} "
- f"crypto_alg {crypto_alg.alg_name} "
- f"local_crypto_key {ckey} remote_crypto_key {ckey} "
- f"{integ} "
- f"local_ip {if1_ip + i * addr_incr} "
- f"remote_ip {if2_ip} "
- f"instance {i}\n"
- )
- tmp_f2.write(
- f"ipsec_tunnel_if_add_del "
- f"local_spi {spi_2 + i} remote_spi {spi_1 + i} "
- f"crypto_alg {crypto_alg.alg_name} "
- f"local_crypto_key {ckey} remote_crypto_key {ckey} "
- f"{integ} "
- f"local_ip {if2_ip} "
- f"remote_ip {if1_ip + i * addr_incr} "
- f"instance {i}\n"
- )
- vat.execute_script(
- tmp_fn1, nodes[u"DUT1"], timeout=1800, json_out=False,
- copy_on_execute=True,
- history=bool(n_tunnels < 100)
- )
- vat.execute_script(
- tmp_fn2, nodes[u"DUT2"], timeout=1800, json_out=False,
- copy_on_execute=True,
- history=bool(n_tunnels < 100)
- )
- os.remove(tmp_fn1)
- os.remove(tmp_fn2)
-
- with open(tmp_fn1, 'w') as tmp_f1, open(tmp_fn2, 'w') as tmp_f2:
- if not existing_tunnels:
- tmp_f2.write(
- f"exec ip route add {if1_ip}/8 via {if2_ip - 1}"
- f" {if2_n}\n"
- )
- for i in range(existing_tunnels, n_tunnels):
- tmp_f1.write(
- f"exec set interface unnumbered ipip{i} use {if1_n}\n"
- f"exec set interface state ipip{i} up\n"
- f"exec ip route add {raddr_ip2 + i}/{mask2} "
- f"via ipip{i}\n"
- )
- tmp_f2.write(
- f"exec set interface unnumbered ipip{i} use {if2_n}\n"
- f"exec set interface state ipip{i} up\n"
- f"exec ip route add {raddr_ip1 + i}/{mask2} "
- f"via ipip{i}\n"
+ ckeys = [bytes()] * existing_tunnels
+ ikeys = [bytes()] * existing_tunnels
+
+ vat = VatExecutor()
+ with open(tmp_fn1, u"w") as tmp_f1:
+ rmac = Topology.get_interface_mac(nodes[u"DUT2"], if2_key) \
+ if u"DUT2" in nodes.keys() \
+ else Topology.get_interface_mac(nodes[u"TG"], if2_key)
+ if not existing_tunnels:
+ tmp_f1.write(
+ f"exec create loopback interface\n"
+ f"exec set interface state loop0 up\n"
+ f"exec set interface ip address {if1_n} "
+ f"{tun_ips[u'ip2'] - 1}/"
+ f"{len(tun_ips[u'ip2'].packed)*8*3//4}\n"
+ f"exec set ip neighbor {if1_n} {tun_ips[u'ip2']} {rmac} "
+ f"static\n"
+ )
+ for i in range(existing_tunnels, n_tunnels):
+ ckeys.append(
+ gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg))
+ )
+ if integ_alg:
+ ikeys.append(
+ gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg))
)
- vat.execute_script(
- tmp_fn1, nodes[u"DUT1"], timeout=1800, json_out=False,
- copy_on_execute=True,
- history=bool(n_tunnels < 100)
- )
- vat.execute_script(
- tmp_fn2, nodes[u"DUT2"], timeout=1800, json_out=False,
- copy_on_execute=True,
- history=bool(n_tunnels < 100)
- )
- os.remove(tmp_fn1)
- os.remove(tmp_fn2)
- return
+ integ = f"integ_alg {integ_alg.alg_name} " \
+ f"local_integ_key {ikeys[i].hex()} " \
+ f"remote_integ_key {ikeys[i].hex()} "
+ else:
+ integ = u""
+ tmp_f1.write(
+ f"exec set interface ip address loop0 "
+ f"{tun_ips[u'ip1'] + i * addr_incr}/32\n"
+ f"ipsec_tunnel_if_add_del "
+ f"local_spi {spi_d[u'spi_1'] + i} "
+ f"remote_spi {spi_d[u'spi_2'] + i} "
+ f"crypto_alg {crypto_alg.alg_name} "
+ f"local_crypto_key {ckeys[i].hex()} "
+ f"remote_crypto_key {ckeys[i].hex()} "
+ f"{integ} "
+ f"local_ip {tun_ips[u'ip1'] + i * addr_incr} "
+ f"remote_ip {tun_ips[u'ip2']} "
+ f"instance {i}\n"
+ )
+ vat.execute_script(
+ tmp_fn1, nodes[u"DUT1"], timeout=1800, json_out=False,
+ copy_on_execute=True,
+ history=bool(n_tunnels < 100)
+ )
+ os.remove(tmp_fn1)
- if not existing_tunnels:
- with PapiSocketExecutor(nodes[u"DUT1"]) as papi_exec:
- # Create loopback interface on DUT1, set it to up state
- cmd1 = u"create_loopback"
- args1 = dict(
- mac_address=0
+ with open(tmp_fn1, 'w') as tmp_f1:
+ for i in range(existing_tunnels, n_tunnels):
+ tmp_f1.write(
+ f"exec set interface unnumbered ipip{i} use {if1_n}\n"
+ f"exec set interface state ipip{i} up\n"
+ f"exec ip route add "
+ f"{raddr_ip2 + i}/{len(raddr_ip2.packed)*8} "
+ f"via ipip{i}\n"
)
- err_msg = f"Failed to create loopback interface " \
- f"on host {nodes[u'DUT1'][u'host']}"
- loop_sw_if_idx = papi_exec.add(cmd1, **args1).\
- get_sw_if_index(err_msg)
- cmd1 = u"sw_interface_set_flags"
- args1 = dict(
- sw_if_index=loop_sw_if_idx,
- flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
+ vat.execute_script(
+ tmp_fn1, nodes[u"DUT1"], timeout=1800, json_out=False,
+ copy_on_execute=True,
+ history=bool(n_tunnels < 100)
+ )
+ os.remove(tmp_fn1)
+
+ return ckeys, ikeys
+
+ @staticmethod
+ def _ipsec_create_tunnel_interfaces_dut2_vat(
+ nodes, tun_ips, if2_key, n_tunnels, crypto_alg, ckeys, integ_alg,
+ ikeys, raddr_ip1, addr_incr, spi_d, existing_tunnels=0):
+ """Create multiple IPsec tunnel interfaces on DUT2 node using VAT.
+
+ :param nodes: VPP nodes to create tunnel interfaces.
+ :param tun_ips: Dictionary with VPP node 1 ipsec tunnel interface
+ IPv4/IPv6 address (ip1) and VPP node 2 ipsec tunnel interface
+ IPv4/IPv6 address (ip2).
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
+ :param n_tunnels: Number of tunnel interfaces to be there at the end.
+ :param crypto_alg: The encryption algorithm name.
+ :param ckeys: List of encryption keys.
+ :param integ_alg: The integrity algorithm name.
+ :param ikeys: List of integrity keys.
+ :param spi_d: Dictionary with SPIs for VPP node 1 and VPP node 2.
+ :param addr_incr: IP / IPv6 address incremental step.
+ :param existing_tunnels: Number of tunnel interfaces before creation.
+ Useful mainly for reconf tests. Default 0.
+ :type nodes: dict
+ :type tun_ips: dict
+ :type if2_key: str
+ :type n_tunnels: int
+ :type crypto_alg: CryptoAlg
+ :type ckeys: list
+ :type integ_alg: IntegAlg
+ :type ikeys: list
+ :type addr_incr: int
+ :type spi_d: dict
+ :type existing_tunnels: int
+ """
+ tmp_fn2 = u"/tmp/ipsec_create_tunnel_dut2.config"
+ if2_n = Topology.get_interface_name(nodes[u"DUT2"], if2_key)
+
+ vat = VatExecutor()
+ with open(tmp_fn2, 'w') as tmp_f2:
+ if not existing_tunnels:
+ tmp_f2.write(
+ f"exec set interface ip address {if2_n}"
+ f" {tun_ips[u'ip2']}/{len(tun_ips[u'ip2'].packed)*8*3/4}\n"
)
- err_msg = f"Failed to set loopback interface state up " \
- f"on host {nodes[u'DUT1'][u'host']}"
- papi_exec.add(cmd1, **args1).get_reply(err_msg)
- # Set IP address on VPP node 1 interface
- cmd1 = u"sw_interface_add_del_address"
- args1 = dict(
- sw_if_index=InterfaceUtil.get_interface_index(
+ for i in range(existing_tunnels, n_tunnels):
+ if integ_alg:
+ integ = f"integ_alg {integ_alg.alg_name} " \
+ f"local_integ_key {ikeys[i].hex()} " \
+ f"remote_integ_key {ikeys[i].hex()} "
+ else:
+ integ = u""
+ tmp_f2.write(
+ f"ipsec_tunnel_if_add_del "
+ f"local_spi {spi_d[u'spi_2'] + i} "
+ f"remote_spi {spi_d[u'spi_1'] + i} "
+ f"crypto_alg {crypto_alg.alg_name} "
+ f"local_crypto_key {ckeys[i].hex()} "
+ f"remote_crypto_key {ckeys[i].hex()} "
+ f"{integ} "
+ f"local_ip {tun_ips[u'ip2']} "
+ f"remote_ip {tun_ips[u'ip1'] + i * addr_incr} "
+ f"instance {i}\n"
+ )
+ vat.execute_script(
+ tmp_fn2, nodes[u"DUT2"], timeout=1800, json_out=False,
+ copy_on_execute=True,
+ history=bool(n_tunnels < 100)
+ )
+ os.remove(tmp_fn2)
+
+ with open(tmp_fn2, 'w') as tmp_f2:
+ if not existing_tunnels:
+ tmp_f2.write(
+ f"exec ip route add {tun_ips[u'ip1']}/8 "
+ f"via {tun_ips[u'ip2'] - 1} {if2_n}\n"
+ )
+ for i in range(existing_tunnels, n_tunnels):
+ tmp_f2.write(
+ f"exec set interface unnumbered ipip{i} use {if2_n}\n"
+ f"exec set interface state ipip{i} up\n"
+ f"exec ip route add "
+ f"{raddr_ip1 + i}/{len(raddr_ip1.packed)*8} "
+ f"via ipip{i}\n"
+ )
+ vat.execute_script(
+ tmp_fn2, nodes[u"DUT2"], timeout=1800, json_out=False,
+ copy_on_execute=True,
+ history=bool(n_tunnels < 100)
+ )
+ os.remove(tmp_fn2)
+
+ @staticmethod
+ def _ipsec_create_loopback_dut1_papi(nodes, tun_ips, if1_key, if2_key):
+ """Create loopback interface and set IP address on VPP node 1 interface
+ using PAPI.
+
+ :param nodes: VPP nodes to create tunnel interfaces.
+ :param tun_ips: Dictionary with VPP node 1 ipsec tunnel interface
+ IPv4/IPv6 address (ip1) and VPP node 2 ipsec tunnel interface
+ IPv4/IPv6 address (ip2).
+ :param if1_key: VPP node 1 interface key from topology file.
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
+ :type nodes: dict
+ :type tun_ips: dict
+ :type if1_key: str
+ :type if2_key: str
+ """
+ with PapiSocketExecutor(nodes[u"DUT1"]) as papi_exec:
+ # Create loopback interface on DUT1, set it to up state
+ cmd = u"create_loopback"
+ args = dict(
+ mac_address=0
+ )
+ err_msg = f"Failed to create loopback interface " \
+ f"on host {nodes[u'DUT1'][u'host']}"
+ loop_sw_if_idx = papi_exec.add(cmd, **args). \
+ get_sw_if_index(err_msg)
+ cmd = u"sw_interface_set_flags"
+ args = dict(
+ sw_if_index=loop_sw_if_idx,
+ flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
+ )
+ err_msg = f"Failed to set loopback interface state up " \
+ f"on host {nodes[u'DUT1'][u'host']}"
+ papi_exec.add(cmd, **args).get_reply(err_msg)
+ # Set IP address on VPP node 1 interface
+ cmd = u"sw_interface_add_del_address"
+ args = dict(
+ sw_if_index=InterfaceUtil.get_interface_index(
+ nodes[u"DUT1"], if1_key
+ ),
+ is_add=True,
+ del_all=False,
+ prefix=IPUtil.create_prefix_object(
+ tun_ips[u"ip2"] - 1, 96 if tun_ips[u"ip2"].version == 6
+ else 24
+ )
+ )
+ err_msg = f"Failed to set IP address on interface {if1_key} " \
+ f"on host {nodes[u'DUT1'][u'host']}"
+ papi_exec.add(cmd, **args).get_reply(err_msg)
+ cmd2 = u"ip_neighbor_add_del"
+ args2 = dict(
+ is_add=1,
+ neighbor=dict(
+ sw_if_index=Topology.get_interface_sw_index(
nodes[u"DUT1"], if1_key
),
- is_add=True,
- del_all=False,
- prefix=IPUtil.create_prefix_object(
- if2_ip - 1, 96 if if2_ip.version == 6 else 24
- )
- )
- err_msg = f"Failed to set IP address on interface {if1_key} " \
- f"on host {nodes[u'DUT1'][u'host']}"
- papi_exec.add(cmd1, **args1).get_reply(err_msg)
- cmd4 = u"ip_neighbor_add_del"
- args4 = dict(
- is_add=1,
- neighbor=dict(
- sw_if_index=Topology.get_interface_sw_index(
- nodes[u"DUT1"], if1_key
- ),
- flags=1,
- mac_address=str(
- Topology.get_interface_mac(nodes[u"DUT2"], if2_key)
- ),
- ip_address=str(ip_address(if2_ip_addr))
- )
+ flags=1,
+ mac_address=str(
+ Topology.get_interface_mac(nodes[u"DUT2"], if2_key)
+ if u"DUT2" in nodes.keys()
+ else Topology.get_interface_mac(
+ nodes[u"TG"], if2_key
+ )
+ ),
+ ip_address=tun_ips[u"ip2"].compressed
)
- err_msg = f"Failed to add IP neighbor on interface {if1_key}"
- papi_exec.add(cmd4, **args4).get_reply(err_msg)
+ )
+ err_msg = f"Failed to add IP neighbor on interface {if1_key}"
+ papi_exec.add(cmd2, **args2).get_reply(err_msg)
+
+ return loop_sw_if_idx
+
+ @staticmethod
+ def _ipsec_create_tunnel_interfaces_dut1_papi(
+ nodes, tun_ips, if1_key, if2_key, n_tunnels, crypto_alg, integ_alg,
+ raddr_ip2, addr_incr, spi_d, existing_tunnels=0):
+ """Create multiple IPsec tunnel interfaces on DUT1 node using PAPI.
+
+ :param nodes: VPP nodes to create tunnel interfaces.
+ :param tun_ips: Dictionary with VPP node 1 ipsec tunnel interface
+ IPv4/IPv6 address (ip1) and VPP node 2 ipsec tunnel interface
+ IPv4/IPv6 address (ip2).
+ :param if1_key: VPP node 1 interface key from topology file.
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
+ :param n_tunnels: Number of tunnel interfaces to be there at the end.
+ :param crypto_alg: The encryption algorithm name.
+ :param integ_alg: The integrity algorithm name.
+ :param raddr_ip2: Policy selector remote IPv4/IPv6 start address for the
+ first tunnel in direction node2->node1.
+ :param spi_d: Dictionary with SPIs for VPP node 1 and VPP node 2.
+ :param addr_incr: IP / IPv6 address incremental step.
+ :param existing_tunnels: Number of tunnel interfaces before creation.
+ Useful mainly for reconf tests. Default 0.
+ :type nodes: dict
+ :type tun_ips: dict
+ :type if1_key: str
+ :type if2_key: str
+ :type n_tunnels: int
+ :type crypto_alg: CryptoAlg
+ :type integ_alg: IntegAlg
+ :type raddr_ip2: IPv4Address or IPv6Address
+ :type addr_incr: int
+ :type spi_d: dict
+ :type existing_tunnels: int
+ """
+ if not existing_tunnels:
+ loop_sw_if_idx = IPsecUtil._ipsec_create_loopback_dut1_papi(
+ nodes, tun_ips, if1_key, if2_key
+ )
else:
- # Executor not open, as InterfaceUtil will open its own.
loop_sw_if_idx = InterfaceUtil.vpp_get_interface_sw_index(
- nodes[u"DUT1"], u"loop0")
- cmd1 = u"sw_interface_add_del_address"
+ nodes[u"DUT1"], u"loop0"
+ )
with PapiSocketExecutor(nodes[u"DUT1"]) as papi_exec:
- # Configure IPsec tunnel interfaces
- args1 = dict(
+ # Configure IP addresses on loop0 interface
+ cmd = u"sw_interface_add_del_address"
+ args = dict(
sw_if_index=loop_sw_if_idx,
is_add=True,
del_all=False,
prefix=None
)
- cmd2 = u"ipsec_tunnel_if_add_del"
- args2 = dict(
+ for i in range(existing_tunnels, n_tunnels):
+ args[u"prefix"] = IPUtil.create_prefix_object(
+ tun_ips[u"ip1"] + i * addr_incr,
+ 128 if tun_ips[u"ip1"].version == 6 else 32
+ )
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ # Configure IPsec tunnel interfaces
+ cmd = u"ipsec_tunnel_if_add_del"
+ args = dict(
is_add=True,
local_ip=None,
remote_ip=None,
remote_integ_key=None,
tx_table_id=0
)
- err_msg = f"Failed to add IPsec tunnel interfaces " \
- f"on host {nodes[u'DUT1'][u'host']}"
ipsec_tunnels = [None] * existing_tunnels
- ckeys = [None] * existing_tunnels
- ikeys = [None] * existing_tunnels
+ ckeys = [bytes()] * existing_tunnels
+ ikeys = [bytes()] * existing_tunnels
for i in range(existing_tunnels, n_tunnels):
ckeys.append(
gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg))
ikeys.append(
gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg))
)
- args1[u"prefix"] = IPUtil.create_prefix_object(
- if1_ip + i * addr_incr, 128 if if1_ip.version == 6 else 32
+ args[u"local_spi"] = spi_d[u"spi_1"] + i
+ args[u"remote_spi"] = spi_d[u"spi_2"] + i
+ args[u"local_ip"] = IPAddress.create_ip_address_object(
+ tun_ips[u"ip1"] + i * addr_incr
)
- args2[u"local_spi"] = spi_1 + i
- args2[u"remote_spi"] = spi_2 + i
- args2[u"local_ip"] = IPAddress.create_ip_address_object(
- if1_ip + i * addr_incr
+ args[u"remote_ip"] = IPAddress.create_ip_address_object(
+ tun_ips[u"ip2"]
)
- args2[u"remote_ip"] = IPAddress.create_ip_address_object(if2_ip)
- args2[u"local_crypto_key_len"] = len(ckeys[i])
- args2[u"local_crypto_key"] = ckeys[i]
- args2[u"remote_crypto_key_len"] = len(ckeys[i])
- args2[u"remote_crypto_key"] = ckeys[i]
+ args[u"local_crypto_key_len"] = len(ckeys[i])
+ args[u"local_crypto_key"] = ckeys[i]
+ args[u"remote_crypto_key_len"] = len(ckeys[i])
+ args[u"remote_crypto_key"] = ckeys[i]
if integ_alg:
- args2[u"local_integ_key_len"] = len(ikeys[i])
- args2[u"local_integ_key"] = ikeys[i]
- args2[u"remote_integ_key_len"] = len(ikeys[i])
- args2[u"remote_integ_key"] = ikeys[i]
- history = bool(not 1 < i < n_tunnels - 2)
- papi_exec.add(cmd1, history=history, **args1).\
- add(cmd2, history=history, **args2)
- replies = papi_exec.get_replies(err_msg)
- for reply in replies:
- if u"sw_if_index" in reply:
- ipsec_tunnels.append(reply[u"sw_if_index"])
- # Configure IP routes
- cmd1 = u"sw_interface_set_unnumbered"
- args1 = dict(
+ args[u"local_integ_key_len"] = len(ikeys[i])
+ args[u"local_integ_key"] = ikeys[i]
+ args[u"remote_integ_key_len"] = len(ikeys[i])
+ args[u"remote_integ_key"] = ikeys[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ err_msg = f"Failed to add IPsec tunnel interfaces on host" \
+ f" {nodes[u'DUT1'][u'host']}"
+ ipsec_tunnels.extend(
+ [
+ reply[u"sw_if_index"]
+ for reply in papi_exec.get_replies(err_msg)
+ if u"sw_if_index" in reply
+ ]
+ )
+ # Configure unnumbered interfaces
+ cmd = u"sw_interface_set_unnumbered"
+ args = dict(
is_add=True,
sw_if_index=InterfaceUtil.get_interface_index(
nodes[u"DUT1"], if1_key
),
unnumbered_sw_if_index=0
)
- cmd2 = u"sw_interface_set_flags"
- args2 = dict(
+ for i in range(existing_tunnels, n_tunnels):
+ args[u"unnumbered_sw_if_index"] = ipsec_tunnels[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ # Set interfaces up
+ cmd = u"sw_interface_set_flags"
+ args = dict(
sw_if_index=0,
flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
)
- cmd3 = u"ip_route_add_del"
- args3 = dict(
+ for i in range(existing_tunnels, n_tunnels):
+ args[u"sw_if_index"] = ipsec_tunnels[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ # Configure IP routes
+ cmd = u"ip_route_add_del"
+ args = dict(
is_add=1,
is_multipath=0,
route=None
)
- err_msg = f"Failed to add IP routes " \
- f"on host {nodes[u'DUT1'][u'host']}"
for i in range(existing_tunnels, n_tunnels):
- args1[u"unnumbered_sw_if_index"] = ipsec_tunnels[i]
- args2[u"sw_if_index"] = ipsec_tunnels[i]
- args3[u"route"] = IPUtil.compose_vpp_route_structure(
+ args[u"route"] = IPUtil.compose_vpp_route_structure(
nodes[u"DUT1"], (raddr_ip2 + i).compressed,
prefix_len=128 if raddr_ip2.version == 6 else 32,
interface=ipsec_tunnels[i]
)
- history = bool(not 1 < i < n_tunnels - 2)
- papi_exec.add(cmd1, history=history, **args1).\
- add(cmd2, history=history, **args2).\
- add(cmd3, history=history, **args3)
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ err_msg = f"Failed to add IP routes on host " \
+ f"{nodes[u'DUT1'][u'host']}"
papi_exec.get_replies(err_msg)
+ return ckeys, ikeys
+
+ @staticmethod
+ def _ipsec_create_tunnel_interfaces_dut2_papi(
+ nodes, tun_ips, if2_key, n_tunnels, crypto_alg, ckeys, integ_alg,
+ ikeys, raddr_ip1, addr_incr, spi_d, existing_tunnels=0):
+ """Create multiple IPsec tunnel interfaces on DUT2 node using PAPI.
+
+ :param nodes: VPP nodes to create tunnel interfaces.
+ :param tun_ips: Dictionary with VPP node 1 ipsec tunnel interface
+ IPv4/IPv6 address (ip1) and VPP node 2 ipsec tunnel interface
+ IPv4/IPv6 address (ip2).
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
+ :param n_tunnels: Number of tunnel interfaces to be there at the end.
+ :param crypto_alg: The encryption algorithm name.
+ :param ckeys: List of encryption keys.
+ :param integ_alg: The integrity algorithm name.
+ :param ikeys: List of integrity keys.
+ :param spi_d: Dictionary with SPIs for VPP node 1 and VPP node 2.
+ :param addr_incr: IP / IPv6 address incremental step.
+ :param existing_tunnels: Number of tunnel interfaces before creation.
+ Useful mainly for reconf tests. Default 0.
+ :type nodes: dict
+ :type tun_ips: dict
+ :type if2_key: str
+ :type n_tunnels: int
+ :type crypto_alg: CryptoAlg
+ :type ckeys: list
+ :type integ_alg: IntegAlg
+ :type ikeys: list
+ :type addr_incr: int
+ :type spi_d: dict
+ :type existing_tunnels: int
+ """
with PapiSocketExecutor(nodes[u"DUT2"]) as papi_exec:
if not existing_tunnels:
# Set IP address on VPP node 2 interface
- cmd1 = u"sw_interface_add_del_address"
- args1 = dict(
+ cmd = u"sw_interface_add_del_address"
+ args = dict(
sw_if_index=InterfaceUtil.get_interface_index(
nodes[u"DUT2"], if2_key
),
is_add=True,
del_all=False,
prefix=IPUtil.create_prefix_object(
- if2_ip, 96 if if2_ip.version == 6 else 24
+ tun_ips[u"ip2"], 96 if tun_ips[u"ip2"].version == 6
+ else 24
)
)
err_msg = f"Failed to set IP address on interface {if2_key} " \
f"on host {nodes[u'DUT2'][u'host']}"
- papi_exec.add(cmd1, **args1).get_reply(err_msg)
+ papi_exec.add(cmd, **args).get_reply(err_msg)
# Configure IPsec tunnel interfaces
- cmd2 = u"ipsec_tunnel_if_add_del"
- args2 = dict(
+ cmd = u"ipsec_tunnel_if_add_del"
+ args = dict(
is_add=True,
- local_ip=IPAddress.create_ip_address_object(if2_ip),
+ local_ip=IPAddress.create_ip_address_object(tun_ips[u"ip2"]),
remote_ip=None,
local_spi=0,
remote_spi=0,
remote_integ_key=None,
tx_table_id=0
)
- err_msg = f"Failed to add IPsec tunnel interfaces " \
- f"on host {nodes[u'DUT2'][u'host']}"
ipsec_tunnels = [None] * existing_tunnels
for i in range(existing_tunnels, n_tunnels):
- args2[u"local_spi"] = spi_2 + i
- args2[u"remote_spi"] = spi_1 + i
- args2[u"local_ip"] = IPAddress.create_ip_address_object(if2_ip)
- args2[u"remote_ip"] = IPAddress.create_ip_address_object(
- if1_ip + i * addr_incr)
- args2[u"local_crypto_key_len"] = len(ckeys[i])
- args2[u"local_crypto_key"] = ckeys[i]
- args2[u"remote_crypto_key_len"] = len(ckeys[i])
- args2[u"remote_crypto_key"] = ckeys[i]
+ args[u"local_spi"] = spi_d[u"spi_2"] + i
+ args[u"remote_spi"] = spi_d[u"spi_1"] + i
+ args[u"local_ip"] = IPAddress.create_ip_address_object(
+ tun_ips[u"ip2"]
+ )
+ args[u"remote_ip"] = IPAddress.create_ip_address_object(
+ tun_ips[u"ip1"] + i * addr_incr
+ )
+ args[u"local_crypto_key_len"] = len(ckeys[i])
+ args[u"local_crypto_key"] = ckeys[i]
+ args[u"remote_crypto_key_len"] = len(ckeys[i])
+ args[u"remote_crypto_key"] = ckeys[i]
if integ_alg:
- args2[u"local_integ_key_len"] = len(ikeys[i])
- args2[u"local_integ_key"] = ikeys[i]
- args2[u"remote_integ_key_len"] = len(ikeys[i])
- args2[u"remote_integ_key"] = ikeys[i]
- history = bool(not 1 < i < n_tunnels - 2)
- papi_exec.add(cmd2, history=history, **args2)
- replies = papi_exec.get_replies(err_msg)
- for reply in replies:
- if u"sw_if_index" in reply:
- ipsec_tunnels.append(reply[u"sw_if_index"])
+ args[u"local_integ_key_len"] = len(ikeys[i])
+ args[u"local_integ_key"] = ikeys[i]
+ args[u"remote_integ_key_len"] = len(ikeys[i])
+ args[u"remote_integ_key"] = ikeys[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ err_msg = f"Failed to add IPsec tunnel interfaces " \
+ f"on host {nodes[u'DUT2'][u'host']}"
+ ipsec_tunnels.extend(
+ [
+ reply[u"sw_if_index"]
+ for reply in papi_exec.get_replies(err_msg)
+ if u"sw_if_index" in reply
+ ]
+ )
if not existing_tunnels:
- # Configure IP routes
- cmd1 = u"ip_route_add_del"
+ # Configure IP route
+ cmd = u"ip_route_add_del"
route = IPUtil.compose_vpp_route_structure(
- nodes[u"DUT2"], if1_ip.compressed,
- prefix_len=32 if if1_ip.version == 6 else 8,
+ nodes[u"DUT2"], tun_ips[u"ip1"].compressed,
+ prefix_len=32 if tun_ips[u"ip1"].version == 6 else 8,
interface=if2_key,
- gateway=(if2_ip - 1).compressed
+ gateway=(tun_ips[u"ip2"] - 1).compressed
)
- args1 = dict(
+ args = dict(
is_add=1,
is_multipath=0,
route=route
)
- papi_exec.add(cmd1, **args1)
- cmd1 = u"sw_interface_set_unnumbered"
- args1 = dict(
+ papi_exec.add(cmd, **args)
+ # Configure unnumbered interfaces
+ cmd = u"sw_interface_set_unnumbered"
+ args = dict(
is_add=True,
sw_if_index=InterfaceUtil.get_interface_index(
nodes[u"DUT2"], if2_key
),
unnumbered_sw_if_index=0
)
- cmd2 = u"sw_interface_set_flags"
- args2 = dict(
+ for i in range(existing_tunnels, n_tunnels):
+ args[u"unnumbered_sw_if_index"] = ipsec_tunnels[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ # Set interfaces up
+ cmd = u"sw_interface_set_flags"
+ args = dict(
sw_if_index=0,
flags=InterfaceStatusFlags.IF_STATUS_API_FLAG_ADMIN_UP.value
)
- cmd3 = u"ip_route_add_del"
- args3 = dict(
+ for i in range(existing_tunnels, n_tunnels):
+ args[u"sw_if_index"] = ipsec_tunnels[i]
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ # Configure IP routes
+ cmd = u"ip_route_add_del"
+ args = dict(
is_add=1,
is_multipath=0,
route=None
)
- err_msg = f"Failed to add IP routes " \
- f"on host {nodes[u'DUT2'][u'host']}"
for i in range(existing_tunnels, n_tunnels):
- args1[u"unnumbered_sw_if_index"] = ipsec_tunnels[i]
- args2[u"sw_if_index"] = ipsec_tunnels[i]
- args3[u"route"] = IPUtil.compose_vpp_route_structure(
+ args[u"route"] = IPUtil.compose_vpp_route_structure(
nodes[u"DUT1"], (raddr_ip1 + i).compressed,
prefix_len=128 if raddr_ip1.version == 6 else 32,
interface=ipsec_tunnels[i]
)
- history = bool(not 1 < i < n_tunnels - 2)
- papi_exec.add(cmd1, history=history, **args1). \
- add(cmd2, history=history, **args2). \
- add(cmd3, history=history, **args3)
+ papi_exec.add(
+ cmd, history=bool(not 1 < i < n_tunnels - 2), **args
+ )
+ err_msg = f"Failed to add IP routes " \
+ f"on host {nodes[u'DUT2'][u'host']}"
papi_exec.get_replies(err_msg)
+ @staticmethod
+ def vpp_ipsec_create_tunnel_interfaces(
+ nodes, tun_if1_ip_addr, tun_if2_ip_addr, if1_key, if2_key,
+ n_tunnels, crypto_alg, integ_alg, raddr_ip1, raddr_ip2, raddr_range,
+ existing_tunnels=0):
+ """Create multiple IPsec tunnel interfaces between two VPP nodes.
+
+ :param nodes: VPP nodes to create tunnel interfaces.
+ :param tun_if1_ip_addr: VPP node 1 ipsec tunnel interface IPv4/IPv6
+ address.
+ :param tun_if2_ip_addr: VPP node 2 ipsec tunnel interface IPv4/IPv6
+ address.
+ :param if1_key: VPP node 1 interface key from topology file.
+ :param if2_key: VPP node 2 / TG node (in case of 2-node topology)
+ interface key from topology file.
+ :param n_tunnels: Number of tunnel interfaces to be there at the end.
+ :param crypto_alg: The encryption algorithm name.
+ :param integ_alg: The integrity algorithm name.
+ :param raddr_ip1: Policy selector remote IPv4/IPv6 start address for the
+ first tunnel in direction node1->node2.
+ :param raddr_ip2: Policy selector remote IPv4/IPv6 start address for the
+ first tunnel in direction node2->node1.
+ :param raddr_range: Mask specifying range of Policy selector Remote
+ IPv4/IPv6 addresses. Valid values are from 1 to 32 in case of IPv4
+ and to 128 in case of IPv6.
+ :param existing_tunnels: Number of tunnel interfaces before creation.
+ Useful mainly for reconf tests. Default 0.
+ :type nodes: dict
+ :type tun_if1_ip_addr: str
+ :type tun_if2_ip_addr: str
+ :type if1_key: str
+ :type if2_key: str
+ :type n_tunnels: int
+ :type crypto_alg: CryptoAlg
+ :type integ_alg: IntegAlg
+ :type raddr_ip1: string
+ :type raddr_ip2: string
+ :type raddr_range: int
+ :type existing_tunnels: int
+ """
+ n_tunnels = int(n_tunnels)
+ existing_tunnels = int(existing_tunnels)
+ spi_d = dict(
+ spi_1=100000,
+ spi_2=200000
+ )
+ tun_ips = dict(
+ ip1=ip_address(tun_if1_ip_addr),
+ ip2=ip_address(tun_if2_ip_addr)
+ )
+ raddr_ip1 = ip_address(raddr_ip1)
+ raddr_ip2 = ip_address(raddr_ip2)
+ addr_incr = 1 << (128 - raddr_range) if tun_ips[u"ip1"].version == 6 \
+ else 1 << (32 - raddr_range)
+
+ if n_tunnels - existing_tunnels > 10:
+ ckeys, ikeys = IPsecUtil._ipsec_create_tunnel_interfaces_dut1_vat(
+ nodes, tun_ips, if1_key, if2_key, n_tunnels, crypto_alg,
+ integ_alg, raddr_ip2, addr_incr, spi_d, existing_tunnels
+ )
+ if u"DUT2" not in nodes.keys():
+ return ckeys[0], ikeys[0], spi_d[u"spi_1"], spi_d[u"spi_2"]
+ IPsecUtil._ipsec_create_tunnel_interfaces_dut2_vat(
+ nodes, tun_ips, if2_key, n_tunnels, crypto_alg, ckeys,
+ integ_alg, ikeys, raddr_ip1, addr_incr, spi_d, existing_tunnels
+ )
+ else:
+ ckeys, ikeys = IPsecUtil._ipsec_create_tunnel_interfaces_dut1_papi(
+ nodes, tun_ips, if1_key, if2_key, n_tunnels, crypto_alg,
+ integ_alg, raddr_ip2, addr_incr, spi_d, existing_tunnels
+ )
+ if u"DUT2" not in nodes.keys():
+ return ckeys[0], ikeys[0], spi_d[u"spi_1"], spi_d[u"spi_2"]
+ IPsecUtil._ipsec_create_tunnel_interfaces_dut2_papi(
+ nodes, tun_ips, if2_key, n_tunnels, crypto_alg, ckeys,
+ integ_alg, ikeys, raddr_ip1, addr_incr, spi_d, existing_tunnels
+ )
+
+ return None, None, None, None
+
@staticmethod
def _create_ipsec_script_files(dut, instances):
"""Create script files for configuring IPsec in containers
@staticmethod
def vpp_ipsec_create_tunnel_interfaces_in_containers(
- nodes, if1_ip_addr, if2_ip_addr, if1_key, if2_key, n_tunnels,
- crypto_alg, integ_alg, raddr_ip1, raddr_ip2, raddr_range,
- n_instances):
+ nodes, if1_ip_addr, if2_ip_addr, n_tunnels, crypto_alg, integ_alg,
+ raddr_ip1, raddr_ip2, raddr_range, n_instances):
"""Create multiple IPsec tunnel interfaces between two VPP nodes.
: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 integ_alg: The integrity algorithm name.
: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 integ_alg: IntegAlg
addr_incr = 1 << (32 - raddr_range)
dut1_scripts = IPsecUtil._create_ipsec_script_files(
- u"DUT1", n_instances)
+ u"DUT1", n_instances
+ )
dut2_scripts = IPsecUtil._create_ipsec_script_files(
- u"DUT2", n_instances)
+ u"DUT2", n_instances
+ )
for cnf in range(0, n_instances):
dut1_scripts[cnf].write(
)
for tnl in range(0, n_tunnels):
- tnl_incr = tnl * addr_incr
cnf = tnl % n_instances
- i = tnl // n_instances
- ckey = gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg)).hex()
+ ckey = getattr(
+ gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg)), u"hex"
+ )
integ = u""
if integ_alg:
- ikey = gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg)).hex()
+ ikey = getattr(
+ gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg)), u"hex"
+ )
integ = (
f"integ-alg {integ_alg.alg_name} "
f"local-integ-key {ikey} "
f"remote-integ-key {ikey} "
)
-
# Configure tunnel end point(s) on left side
dut1_scripts[cnf].write(
u"set interface ip address loop0 "
- f"{ip_address(if1_ip_addr) + tnl_incr}/32\n"
+ f"{ip_address(if1_ip_addr) + tnl * addr_incr}/32\n"
f"create ipsec tunnel "
- f"local-ip {ip_address(if1_ip_addr) + tnl_incr} "
+ f"local-ip {ip_address(if1_ip_addr) + tnl * addr_incr} "
f"local-spi {spi_1 + tnl} "
f"remote-ip {ip_address(if2_ip_addr) + cnf} "
f"remote-spi {spi_2 + tnl} "
f"crypto-alg {crypto_alg.alg_name} "
f"local-crypto-key {ckey} "
f"remote-crypto-key {ckey} "
- f"instance {i} "
+ f"instance {tnl // n_instances} "
f"salt 0x0 "
f"{integ} \n"
- f"set interface unnumbered ipip{i} use loop0\n"
- f"set interface state ipip{i} up\n"
- f"ip route add {ip_address(raddr_ip2)+tnl}/32 via ipip{i}\n\n"
+ f"set interface unnumbered ipip{tnl // n_instances} use loop0\n"
+ f"set interface state ipip{tnl // n_instances} up\n"
+ f"ip route add {ip_address(raddr_ip2)+tnl}/32 "
+ f"via ipip{tnl // n_instances}\n\n"
)
-
# Configure tunnel end point(s) on right side
dut2_scripts[cnf].write(
f"set ip neighbor memif1/{cnf + 1} "
- f"{ip_address(if1_ip_addr) + tnl_incr} "
+ f"{ip_address(if1_ip_addr) + tnl * addr_incr} "
f"02:02:00:00:{17:02X}:{cnf:02X} static\n"
f"create ipsec tunnel local-ip {ip_address(if2_ip_addr) + cnf} "
f"local-spi {spi_2 + tnl} "
- f"remote-ip {ip_address(if1_ip_addr) + tnl_incr} "
+ f"remote-ip {ip_address(if1_ip_addr) + tnl * addr_incr} "
f"remote-spi {spi_1 + tnl} "
f"crypto-alg {crypto_alg.alg_name} "
f"local-crypto-key {ckey} "
f"remote-crypto-key {ckey} "
- f"instance {i} "
+ f"instance {tnl // n_instances} "
f"salt 0x0 "
f"{integ}\n"
- f"set interface unnumbered ipip{i} use memif1/{cnf + 1}\n"
- f"set interface state ipip{i} up\n"
- f"ip route add {ip_address(raddr_ip1) + tnl}/32 via ipip{i}\n\n"
+ f"set interface unnumbered ipip{tnl // n_instances} "
+ f"use memif1/{cnf + 1}\n"
+ f"set interface state ipip{tnl // n_instances} up\n"
+ f"ip route add {ip_address(raddr_ip1) + tnl}/32 "
+ f"via ipip{tnl // n_instances}\n\n"
)
IPsecUtil._close_and_copy_ipsec_script_files(
--- /dev/null
+#!/usr/bin/env python3
+
+# Copyright (c) 2020 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.
+
+"""Traffic script for IPsec verification."""
+
+import sys
+
+from ipaddress import ip_address
+from scapy.layers.inet import IP
+from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
+from scapy.layers.ipsec import SecurityAssociation, ESP
+from scapy.layers.l2 import Ether
+from scapy.packet import Raw
+
+from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
+from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
+
+
+def check_ipsec(
+ pkt_recv, ip_layer, src_mac, dst_mac, src_tun, dst_tun, src_ip, dst_ip,
+ sa_in):
+ """Check received IPsec packet.
+
+ :param pkt_recv: Received packet to verify.
+ :param ip_layer: Scapy IP layer.
+ :param src_mac: Source MAC address.
+ :param dst_mac: Destination MAC address.
+ :param src_tun: IPsec tunnel source address.
+ :param dst_tun: IPsec tunnel destination address.
+ :param src_ip: Source IP/IPv6 address of original IP/IPv6 packet.
+ :param dst_ip: Destination IP/IPv6 address of original IP/IPv6 packet.
+ :param sa_in: IPsec SA for packet decryption.
+ :type pkt_recv: scapy.Ether
+ :type ip_layer: scapy.layers.inet.IP or scapy.layers.inet6.IPv6
+ :type src_mac: str
+ :type dst_mac: str
+ :type src_tun: str
+ :type dst_tun: str
+ :type src_ip: str
+ :type dst_ip: str
+ :type sa_in: scapy.layers.ipsec.SecurityAssociation
+ :raises RuntimeError: If received packet is invalid.
+ """
+ if pkt_recv[Ether].src != src_mac:
+ raise RuntimeError(
+ f"Received frame has invalid source MAC address: "
+ f"{pkt_recv[Ether].src} should be: {src_mac}"
+ )
+
+ if pkt_recv[Ether].dst != dst_mac:
+ raise RuntimeError(
+ f"Received frame has invalid destination MAC address: "
+ f"{pkt_recv[Ether].dst} should be: {dst_mac}"
+ )
+
+ if not pkt_recv.haslayer(ip_layer):
+ raise RuntimeError(
+ f"Not an {ip_layer.name} packet received: {pkt_recv!r}"
+ )
+
+ if pkt_recv[ip_layer].src != src_tun:
+ raise RuntimeError(
+ f"Received packet has invalid source address: "
+ f"{pkt_recv[ip_layer].src} should be: {src_tun}"
+ )
+
+ if pkt_recv[ip_layer].dst != dst_tun:
+ raise RuntimeError(
+ f"Received packet has invalid destination address: "
+ f"{pkt_recv[ip_layer].dst} should be: {dst_tun}"
+ )
+
+ if not pkt_recv.haslayer(ESP):
+ raise RuntimeError(f"Not an ESP packet received: {pkt_recv!r}")
+
+ ip_pkt = pkt_recv[ip_layer]
+ d_pkt = sa_in.decrypt(ip_pkt)
+
+ if d_pkt[ip_layer].dst != dst_ip:
+ raise RuntimeError(
+ f"Decrypted packet has invalid destination address: "
+ f"{d_pkt[ip_layer].dst} should be: {dst_ip}"
+ )
+
+ if d_pkt[ip_layer].src != src_ip:
+ raise RuntimeError(
+ f"Decrypted packet has invalid source address: "
+ f"{d_pkt[ip_layer].src} should be: {src_ip}"
+ )
+
+ if ip_layer == IP and d_pkt[ip_layer].proto != 61:
+ raise RuntimeError(
+ f"Decrypted packet has invalid IP protocol: "
+ f"{d_pkt[ip_layer].proto} should be: 61"
+ )
+
+
+def check_ip(pkt_recv, ip_layer, src_mac, dst_mac, src_ip, dst_ip):
+ """Check received IP/IPv6 packet.
+
+ :param pkt_recv: Received packet to verify.
+ :param ip_layer: Scapy IP layer.
+ :param src_mac: Source MAC address.
+ :param dst_mac: Destination MAC address.
+ :param src_ip: Source IP/IPv6 address.
+ :param dst_ip: Destination IP/IPv6 address.
+ :type pkt_recv: scapy.Ether
+ :type ip_layer: scapy.layers.inet.IP or scapy.layers.inet6.IPv6
+ :type src_mac: str
+ :type dst_mac: str
+ :type src_ip: str
+ :type dst_ip: str
+ :raises RuntimeError: If received packet is invalid.
+ """
+ if pkt_recv[Ether].src != src_mac:
+ raise RuntimeError(
+ f"Received frame has invalid source MAC address: "
+ f"{pkt_recv[Ether].src} should be: {src_mac}"
+ )
+
+ if pkt_recv[Ether].dst != dst_mac:
+ raise RuntimeError(
+ f"Received frame has invalid destination MAC address: "
+ f"{pkt_recv[Ether].dst} should be: {dst_mac}"
+ )
+
+ if not pkt_recv.haslayer(ip_layer):
+ raise RuntimeError(
+ f"Not an {ip_layer.name} packet received: {pkt_recv!r}"
+ )
+
+ if pkt_recv[ip_layer].dst != dst_ip:
+ raise RuntimeError(
+ f"Received packet has invalid destination address: "
+ f"{pkt_recv[ip_layer.name].dst} should be: {dst_ip}"
+ )
+
+ if pkt_recv[ip_layer].src != src_ip:
+ raise RuntimeError(
+ f"Received packet has invalid destination address: "
+ f"{pkt_recv[ip_layer.name].dst} should be: {src_ip}"
+ )
+
+ if ip_layer == IP and pkt_recv[ip_layer].proto != 61:
+ raise RuntimeError(
+ f"Received packet has invalid IP protocol: "
+ f"{pkt_recv[ip_layer].proto} should be: 61"
+ )
+
+
+def main():
+ """Send and receive IPsec packet."""
+
+ args = TrafficScriptArg(
+ [
+ u"tx_src_mac", u"tx_dst_mac", u"rx_src_mac", u"rx_dst_mac",
+ u"src_ip", u"dst_ip", u"src_tun", u"dst_tun", u"crypto_alg",
+ u"crypto_key", u"integ_alg", u"integ_key", u"l_spi", u"r_spi"
+ ]
+ )
+
+ tx_txq = TxQueue(args.get_arg(u"tx_if"))
+ tx_rxq = RxQueue(args.get_arg(u"tx_if"))
+ rx_txq = TxQueue(args.get_arg(u"rx_if"))
+ rx_rxq = RxQueue(args.get_arg(u"rx_if"))
+
+ tx_src_mac = args.get_arg(u"tx_src_mac")
+ tx_dst_mac = args.get_arg(u"tx_dst_mac")
+ rx_src_mac = args.get_arg(u"rx_src_mac")
+ rx_dst_mac = args.get_arg(u"rx_dst_mac")
+ src_ip = args.get_arg(u"src_ip")
+ dst_ip = args.get_arg(u"dst_ip")
+ src_tun = args.get_arg(u"src_tun")
+ dst_tun = args.get_arg(u"dst_tun")
+ crypto_alg = args.get_arg(u"crypto_alg")
+ crypto_key = args.get_arg(u"crypto_key")
+ integ_alg = args.get_arg(u"integ_alg")
+ integ_key = args.get_arg(u"integ_key")
+ l_spi = int(args.get_arg(u"l_spi"))
+ r_spi = int(args.get_arg(u"r_spi"))
+
+ ip_layer = IP if ip_address(src_ip).version == 4 else IPv6
+ ip_pkt = ip_layer(src=src_ip, dst=dst_ip, proto=61) if ip_layer == IP \
+ else ip_layer(src=src_ip, dst=dst_ip)
+
+ tunnel_in = ip_layer(src=src_tun, dst=dst_tun)
+ tunnel_out = ip_layer(src=dst_tun, dst=src_tun)
+
+ sa_in = SecurityAssociation(
+ ESP, spi=l_spi, crypt_algo=crypto_alg,
+ crypt_key=crypto_key.encode(encoding=u"utf-8"), auth_algo=integ_alg,
+ auth_key=integ_key.encode(encoding=u"utf-8"),
+ tunnel_header=tunnel_in
+ )
+
+ sa_out = SecurityAssociation(
+ ESP, spi=r_spi, crypt_algo=crypto_alg,
+ crypt_key=crypto_key.encode(encoding=u"utf-8"), auth_algo=integ_alg,
+ auth_key=integ_key.encode(encoding=u"utf-8"),
+ tunnel_header=tunnel_out
+ )
+
+ sent_packets = list()
+ tx_pkt_send = (Ether(src=tx_src_mac, dst=tx_dst_mac) / ip_pkt)
+ tx_pkt_send /= Raw()
+ size_limit = 78 if ip_layer == IPv6 else 64
+ if len(tx_pkt_send) < size_limit:
+ tx_pkt_send[Raw].load += (b"\0" * (size_limit - len(tx_pkt_send)))
+ sent_packets.append(tx_pkt_send)
+ tx_txq.send(tx_pkt_send)
+
+ while True:
+ rx_pkt_recv = rx_rxq.recv(2)
+
+ if rx_pkt_recv is None:
+ raise RuntimeError(f"{ip_layer.name} packet Rx timeout")
+
+ if rx_pkt_recv.haslayer(ICMPv6ND_NS):
+ # read another packet in the queue if the current one is ICMPv6ND_NS
+ continue
+ else:
+ # otherwise process the current packet
+ break
+
+ check_ipsec(
+ rx_pkt_recv, ip_layer, rx_src_mac, rx_dst_mac, src_tun, dst_tun, src_ip,
+ dst_ip, sa_in
+ )
+
+ ip_pkt = ip_layer(src=dst_ip, dst=src_ip, proto=61) if ip_layer == IP \
+ else ip_layer(src=dst_ip, dst=src_ip)
+ ip_pkt /= Raw()
+ if len(ip_pkt) < (size_limit - 14):
+ ip_pkt[Raw].load += (b"\0" * (size_limit - 14 - len(ip_pkt)))
+ e_pkt = sa_out.encrypt(ip_pkt)
+ rx_pkt_send = (Ether(src=rx_dst_mac, dst=rx_src_mac) /
+ e_pkt)
+ rx_txq.send(rx_pkt_send)
+
+ while True:
+ tx_pkt_recv = tx_rxq.recv(2, ignore=sent_packets)
+
+ if tx_pkt_recv is None:
+ raise RuntimeError(f"{ip_layer.name} packet Rx timeout")
+
+ if tx_pkt_recv.haslayer(ICMPv6ND_NS):
+ # read another packet in the queue if the current one is ICMPv6ND_NS
+ continue
+ else:
+ # otherwise process the current packet
+ break
+
+ check_ip(tx_pkt_recv, ip_layer, tx_dst_mac, tx_src_mac, dst_ip, src_ip)
+
+ sys.exit(0)
+
+
+if __name__ == u"__main__":
+ main()
Code Review / csit.git / commitdiff