from robot.libraries.BuiltIn import BuiltIn
from resources.libraries.python.Constants import Constants
+from resources.libraries.python.enum_util import get_enum_instance
from resources.libraries.python.IncrementUtil import ObjIncrement
from resources.libraries.python.InterfaceUtil import (
InterfaceUtil,
)
-class PolicyAction(Enum):
- """Policy actions."""
+# TODO: Introduce a metaclass that adds .find and .InputType automatically?
+class IpsecSpdAction(Enum):
+ """IPsec SPD actions.
- BYPASS = ("bypass", 0)
+ Mirroring VPP: src/vnet/ipsec/ipsec_types.api enum ipsec_spd_action.
+ """
+
+ BYPASS = NONE = ("bypass", 0)
DISCARD = ("discard", 1)
+ RESOLVE = ("resolve", 2)
PROTECT = ("protect", 3)
- def __init__(self, policy_name: str, policy_int_repr: int):
- self.policy_name = policy_name
- self.policy_int_repr = policy_int_repr
+ def __init__(self, action_name: str, action_int_repr: int):
+ self.action_name = action_name
+ self.action_int_repr = action_int_repr
def __str__(self) -> str:
- return self.policy_name
+ return self.action_name
def __int__(self) -> int:
- return self.policy_int_repr
+ return self.action_int_repr
class CryptoAlg(Enum):
"""Encryption algorithms."""
+ NONE = ("none", 0, "none", 0)
AES_CBC_128 = ("aes-cbc-128", 1, "AES-CBC", 16)
AES_CBC_256 = ("aes-cbc-256", 3, "AES-CBC", 32)
AES_GCM_128 = ("aes-gcm-128", 7, "AES-GCM", 16)
self.scapy_name = scapy_name
self.key_len = key_len
+ # TODO: Investigate if __int__ works with PAPI. It was not enough for "if".
+ def __bool__(self):
+ """A shorthand to enable "if crypto_alg:" constructs."""
+ return self.alg_int_repr != 0
+
class IntegAlg(Enum):
"""Integrity algorithm."""
+ NONE = ("none", 0, "none", 0)
SHA_256_128 = ("sha-256-128", 4, "SHA2-256-128", 32)
SHA_512_256 = ("sha-512-256", 6, "SHA2-512-256", 64)
self.scapy_name = scapy_name
self.key_len = key_len
+ def __bool__(self):
+ """A shorthand to enable "if integ_alg:" constructs."""
+ return self.alg_int_repr != 0
+
+# TODO: Base on Enum, so str values can be defined as in alg enums?
class IPsecProto(IntEnum):
- """IPsec protocol."""
+ """IPsec protocol.
+
+ Mirroring VPP: src/vnet/ipsec/ipsec_types.api enum ipsec_proto.
+ """
+
+ ESP = 50
+ AH = 51
+ NONE = 255
- IPSEC_API_PROTO_ESP = 50
- IPSEC_API_PROTO_AH = 51
+ def __str__(self) -> str:
+ """Return string suitable for CLI commands.
+
+ None is not supported.
+ :returns: Lowercase name of the proto.
+ :rtype: str
+ :raises: ValueError if the numeric value is not recognized.
+ """
+ num = int(self)
+ if num == 50:
+ return "esp"
+ if num == 51:
+ return "ah"
+ raise ValueError(f"String form not defined for IPsecProto {num}")
+
+# The rest of enums do not appear outside this file, so no no change needed yet.
class IPsecSadFlags(IntEnum):
"""IPsec Security Association Database flags."""
- IPSEC_API_SAD_FLAG_NONE = 0
+ IPSEC_API_SAD_FLAG_NONE = NONE = 0
# Enable extended sequence numbers
IPSEC_API_SAD_FLAG_USE_ESN = 0x01
# Enable Anti - replay
class TunnelEncpaDecapFlags(IntEnum):
"""Flags controlling tunnel behaviour."""
- TUNNEL_API_ENCAP_DECAP_FLAG_NONE = 0
+ TUNNEL_API_ENCAP_DECAP_FLAG_NONE = NONE = 0
# at encap, copy the DF bit of the payload into the tunnel header
TUNNEL_API_ENCAP_DECAP_FLAG_ENCAP_COPY_DF = 1
# at encap, set the DF bit in the tunnel header
"""Tunnel modes."""
# point-to-point
- TUNNEL_API_MODE_P2P = 0
+ TUNNEL_API_MODE_P2P = NONE = 0
# multi-point
TUNNEL_API_MODE_MP = 1
-class IPsecUtil:
- """IPsec utilities."""
-
- @staticmethod
- def policy_action_bypass() -> PolicyAction:
- """Return policy action bypass.
-
- :returns: PolicyAction enum BYPASS object.
- :rtype: PolicyAction
- """
- return PolicyAction.BYPASS
-
- @staticmethod
- def policy_action_discard() -> PolicyAction:
- """Return policy action discard.
-
- :returns: PolicyAction enum DISCARD object.
- :rtype: PolicyAction
- """
- return PolicyAction.DISCARD
-
- @staticmethod
- def policy_action_protect() -> PolicyAction:
- """Return policy action protect.
+# Derived types for type hints, based on capabilities of get_enum_instance.
+IpsecSpdAction.InputType = Union[IpsecSpdAction, str, None]
+CryptoAlg.InputType = Union[CryptoAlg, str, None]
+IntegAlg.InputType = Union[IntegAlg, str, None]
+IPsecProto.InputType = Union[IPsecProto, str, int, None]
+# TODO: Introduce a metaclass that adds .find and .InputType automatically?
- :returns: PolicyAction enum PROTECT object.
- :rtype: PolicyAction
- """
- return PolicyAction.PROTECT
- @staticmethod
- def crypto_alg_aes_cbc_128() -> CryptoAlg:
- """Return encryption algorithm aes-cbc-128.
-
- :returns: CryptoAlg enum AES_CBC_128 object.
- :rtype: CryptoAlg
- """
- return CryptoAlg.AES_CBC_128
-
- @staticmethod
- def crypto_alg_aes_cbc_256() -> CryptoAlg:
- """Return encryption algorithm aes-cbc-256.
-
- :returns: CryptoAlg enum AES_CBC_256 object.
- :rtype: CryptoAlg
- """
- return CryptoAlg.AES_CBC_256
-
- @staticmethod
- def crypto_alg_aes_gcm_128() -> CryptoAlg:
- """Return encryption algorithm aes-gcm-128.
-
- :returns: CryptoAlg enum AES_GCM_128 object.
- :rtype: CryptoAlg
- """
- return CryptoAlg.AES_GCM_128
-
- @staticmethod
- def crypto_alg_aes_gcm_256() -> CryptoAlg:
- """Return encryption algorithm aes-gcm-256.
+class IPsecUtil:
+ """IPsec utilities."""
- :returns: CryptoAlg enum AES_GCM_128 object.
- :rtype: CryptoAlg
- """
- return CryptoAlg.AES_GCM_256
+ # The following 4 methods are Python one-liners,
+ # but they are useful when called as a Robot keyword.
@staticmethod
- def get_crypto_alg_key_len(crypto_alg: CryptoAlg) -> int:
+ def get_crypto_alg_key_len(crypto_alg: CryptoAlg.InputType) -> int:
"""Return encryption algorithm key length.
+ This is a Python one-liner, but useful when called as a Robot keyword.
+
:param crypto_alg: Encryption algorithm.
- :type crypto_alg: CryptoAlg
+ :type crypto_alg: CryptoAlg.InputType
:returns: Key length.
:rtype: int
"""
- return crypto_alg.key_len
+ return get_enum_instance(CryptoAlg, crypto_alg).key_len
@staticmethod
- def get_crypto_alg_scapy_name(crypto_alg: CryptoAlg) -> str:
+ def get_crypto_alg_scapy_name(crypto_alg: CryptoAlg.InputType) -> str:
"""Return encryption algorithm scapy name.
+ This is a Python one-liner, but useful when called as a Robot keyword.
+
:param crypto_alg: Encryption algorithm.
- :type crypto_alg: CryptoAlg
+ :type crypto_alg: CryptoAlg.InputType
:returns: Algorithm scapy name.
:rtype: str
"""
- return crypto_alg.scapy_name
+ return get_enum_instance(CryptoAlg, crypto_alg).scapy_name
+ # The below to keywords differ only by enum type conversion from str.
@staticmethod
- def integ_alg_sha_256_128() -> IntegAlg:
- """Return integrity algorithm SHA-256-128.
-
- :returns: IntegAlg enum SHA_256_128 object.
- :rtype: IntegAlg
- """
- return IntegAlg.SHA_256_128
-
- @staticmethod
- def integ_alg_sha_512_256() -> IntegAlg:
- """Return integrity algorithm SHA-512-256.
-
- :returns: IntegAlg enum SHA_512_256 object.
- :rtype: IntegAlg
- """
- return IntegAlg.SHA_512_256
-
- @staticmethod
- def get_integ_alg_key_len(integ_alg: Optional[IntegAlg]) -> int:
+ def get_integ_alg_key_len(integ_alg: IntegAlg.InputType) -> int:
"""Return integrity algorithm key length.
- None argument is accepted, returning zero.
-
:param integ_alg: Integrity algorithm.
- :type integ_alg: Optional[IntegAlg]
+ :type integ_alg: IntegAlg.InputType
:returns: Key length.
:rtype: int
"""
- return 0 if integ_alg is None else integ_alg.key_len
+ return get_enum_instance(IntegAlg, integ_alg).key_len
@staticmethod
- def get_integ_alg_scapy_name(integ_alg: Optional[IntegAlg]) -> str:
+ def get_integ_alg_scapy_name(integ_alg: IntegAlg.InputType) -> str:
"""Return integrity algorithm scapy name.
:param integ_alg: Integrity algorithm.
- :type integ_alg: IntegAlg
+ :type integ_alg: IntegAlg.InputType
:returns: Algorithm scapy name.
:rtype: str
"""
- return integ_alg.scapy_name
-
- @staticmethod
- def ipsec_proto_esp() -> int:
- """Return IPSec protocol ESP.
-
- :returns: IPsecProto enum ESP object.
- :rtype: IPsecProto
- """
- return int(IPsecProto.IPSEC_API_PROTO_ESP)
-
- @staticmethod
- def ipsec_proto_ah() -> int:
- """Return IPSec protocol AH.
-
- :returns: IPsecProto enum AH object.
- :rtype: IPsecProto
- """
- return int(IPsecProto.IPSEC_API_PROTO_AH)
+ return get_enum_instance(IntegAlg, integ_alg).scapy_name
@staticmethod
def vpp_ipsec_select_backend(
- node: dict, protocol: int, index: int = 1
+ node: dict, proto: IPsecProto.InputType, index: int = 1
) -> None:
"""Select IPsec backend.
:param node: VPP node to select IPsec backend on.
- :param protocol: IPsec protocol.
+ :param proto: IPsec protocol.
:param index: Backend index.
:type node: dict
- :type protocol: IPsecProto
+ :type proto: IPsecProto.InputType
:type index: int
:raises RuntimeError: If failed to select IPsec backend or if no API
reply received.
"""
+ proto = get_enum_instance(IPsecProto, proto)
cmd = "ipsec_select_backend"
err_msg = f"Failed to select IPsec backend on host {node['host']}"
- args = dict(protocol=protocol, index=index)
+ args = dict(protocol=proto, index=index)
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args).get_reply(err_msg)
node: dict,
sad_id: int,
spi: int,
- crypto_alg: CryptoAlg,
- crypto_key: str,
- integ_alg: Optional[IntegAlg] = None,
+ crypto_alg: CryptoAlg.InputType = None,
+ crypto_key: str = "",
+ integ_alg: IntegAlg.InputType = None,
integ_key: str = "",
tunnel_src: Optional[str] = None,
tunnel_dst: Optional[str] = None,
:type node: dict
:type sad_id: int
:type spi: int
- :type crypto_alg: CryptoAlg
+ :type crypto_alg: CryptoAlg.InputType
:type crypto_key: str
- :type integ_alg: Optional[IntegAlg]
+ :type integ_alg: IntegAlg.InputType
:type integ_key: str
:type tunnel_src: Optional[str]
:type tunnel_dst: Optional[str]
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
if isinstance(crypto_key, str):
crypto_key = crypto_key.encode(encoding="utf-8")
if isinstance(integ_key, str):
spi=int(spi),
crypto_algorithm=crypto_alg.alg_int_repr,
crypto_key=ckey,
- integrity_algorithm=integ_alg.alg_int_repr if integ_alg else 0,
+ integrity_algorithm=integ_alg.alg_int_repr,
integrity_key=ikey,
flags=flags,
tunnel=dict(
),
dscp=int(IpDscp.IP_API_DSCP_CS0),
),
- protocol=int(IPsecProto.IPSEC_API_PROTO_ESP),
+ protocol=IPsecProto.ESP,
udp_src_port=IPSEC_UDP_PORT_DEFAULT,
udp_dst_port=IPSEC_UDP_PORT_DEFAULT,
anti_replay_window_size=IPSEC_REPLAY_WINDOW_DEFAULT,
n_entries: int,
sad_id: int,
spi: int,
- crypto_alg: CryptoAlg,
- crypto_key: str,
- integ_alg: Optional[IntegAlg] = None,
+ crypto_alg: CryptoAlg.InputType = None,
+ crypto_key: str = "",
+ integ_alg: IntegAlg.InputType = None,
integ_key: str = "",
tunnel_src: Optional[str] = None,
tunnel_dst: Optional[str] = None,
:type n_entries: int
:type sad_id: int
:type spi: int
- :type crypto_alg: CryptoAlg
+ :type crypto_alg: CryptoAlg.InputType
:type crypto_key: str
- :type integ_alg: Optional[IntegAlg]
+ :type integ_alg: IntegAlg.InputType
:type integ_key: str
:type tunnel_src: Optional[str]
:type tunnel_dst: Optional[str]
:type tunnel_addr_incr: bool
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
if isinstance(crypto_key, str):
crypto_key = crypto_key.encode(encoding="utf-8")
if isinstance(integ_key, str):
spi=int(spi),
crypto_algorithm=crypto_alg.alg_int_repr,
crypto_key=ckey,
- integrity_algorithm=integ_alg.alg_int_repr if integ_alg else 0,
+ integrity_algorithm=integ_alg.alg_int_repr,
integrity_key=ikey,
flags=flags,
tunnel=dict(
),
dscp=int(IpDscp.IP_API_DSCP_CS0),
),
- protocol=int(IPsecProto.IPSEC_API_PROTO_ESP),
+ protocol=IPsecProto.ESP,
udp_src_port=IPSEC_UDP_PORT_DEFAULT,
udp_dst_port=IPSEC_UDP_PORT_DEFAULT,
anti_replay_window_size=IPSEC_REPLAY_WINDOW_DEFAULT,
entry_amount: int,
local_addr_range: Union[str, IPv4Address, IPv6Address],
remote_addr_range: Union[str, IPv4Address, IPv6Address],
- action: PolicyAction = PolicyAction.BYPASS,
+ action: IpsecSpdAction.InputType = IpsecSpdAction.BYPASS,
inbound: bool = False,
bidirectional: bool = True,
) -> None:
:param remote_addr_range: Matching remote address range in
direction 1 in format IP/prefix or IP/mask. If no mask is
provided, it's considered to be /32.
- :param action: Policy action.
+ :param action: IPsec SPD action.
:param inbound: If True policy is for inbound traffic, otherwise
outbound.
:param bidirectional: When True, will create SPDs in both directions
Union[str, IPv4Address, IPv6Address]
:type remote_addr_range:
Union[str, IPv4Address, IPv6Address]
- :type action: PolicyAction
+ :type action: IpsecSpdAction.InputType
:type inbound: bool
:type bidirectional: bool
- :raises NotImplementedError: When the action is PolicyAction.PROTECT.
+ :raises NotImplementedError: When the action is IpsecSpdAction.PROTECT.
"""
-
- if action == PolicyAction.PROTECT:
- raise NotImplementedError("Policy action PROTECT is not supported.")
+ action = get_enum_instance(IpsecSpdAction, action)
+ if action == IpsecSpdAction.PROTECT:
+ raise NotImplementedError(
+ "IPsec SPD action PROTECT is not supported."
+ )
spd_id_dir1 = 1
spd_id_dir2 = 2
executor: PapiSocketExecutor,
spd_id: int,
priority: int,
- action: PolicyAction,
+ action: IpsecSpdAction.InputType,
inbound: bool = True,
sa_id: Optional[int] = None,
- proto: Optional[int] = None,
+ proto: IPsecProto.InputType = None,
laddr_range: Optional[str] = None,
raddr_range: Optional[str] = None,
lport_range: Optional[str] = None,
:param executor: Open PAPI executor (async handling) to add commands to.
:param spd_id: SPD ID to add entry on.
:param priority: SPD entry priority, higher number = higher priority.
- :param action: Policy action.
+ :param action: IPsec SPD action.
:param inbound: If True policy is for inbound traffic, otherwise
outbound.
- :param sa_id: SAD entry ID for action PolicyAction.PROTECT.
+ :param sa_id: SAD entry ID for action IpsecSpdAction.PROTECT.
:param proto: Policy selector next layer protocol number.
:param laddr_range: Policy selector local IPv4 or IPv6 address range
in format IP/prefix or IP/mask. If no mask is provided,
:type executor: PapiSocketExecutor
:type spd_id: int
:type priority: int
- :type action: PolicyAction
+ :type action: IpsecSpdAction.InputType
:type inbound: bool
:type sa_id: Optional[int]
- :type proto: Optional[int]
+ :type proto: IPsecProto.InputType
:type laddr_range: Optional[str]
:type raddr_range: Optional[str]
:type lport_range: Optional[str]
:type rport_range: Optional[str]
:type is_ipv6: bool
"""
+ action = get_enum_instance(IpsecSpdAction, action)
+ proto = get_enum_instance(IPsecProto, proto)
if laddr_range is None:
laddr_range = "::/0" if is_ipv6 else "0.0.0.0/0"
is_outbound=not inbound,
sa_id=int(sa_id) if sa_id else 0,
policy=int(action),
- protocol=255 if proto is None else int(proto),
+ protocol=proto,
remote_address_start=IPAddress.create_ip_address_object(
remote_net.network_address
),
node: dict,
spd_id: int,
priority: int,
- action: PolicyAction,
+ action: IpsecSpdAction.InputType,
inbound: bool = True,
sa_id: Optional[int] = None,
- proto: Optional[int] = None,
+ proto: IPsecProto.InputType = None,
laddr_range: Optional[str] = None,
raddr_range: Optional[str] = None,
lport_range: Optional[str] = None,
:param node: VPP node to add SPD entry on.
:param spd_id: SPD ID to add entry on.
:param priority: SPD entry priority, higher number = higher priority.
- :param action: Policy action.
+ :param action: IPsec SPD action.
:param inbound: If True policy is for inbound traffic, otherwise
outbound.
- :param sa_id: SAD entry ID for action PolicyAction.PROTECT.
+ :param sa_id: SAD entry ID for action IpsecSpdAction.PROTECT.
:param proto: Policy selector next layer protocol number.
:param laddr_range: Policy selector local IPv4 or IPv6 address range
in format IP/prefix or IP/mask. If no mask is provided,
:type node: dict
:type spd_id: int
:type priority: int
- :type action: PolicyAction
+ :type action: IpsecSpdAction.InputType
:type inbound: bool
:type sa_id: Optional[int]
- :type proto: Optional[int]
+ :type proto: IPsecProto.InputType
:type laddr_range: Optional[str]
:type raddr_range: Optional[str]
:type lport_range: Optional[str]
:type rport_range: Optional[str]
:type is_ipv6: bool
"""
+ action = get_enum_instance(IpsecSpdAction, action)
+ proto = get_enum_instance(IPsecProto, proto)
err_msg = (
"Failed to add entry to Security Policy Database"
f" {spd_id} on host {node['host']}"
n_entries: int,
spd_id: int,
priority: Optional[ObjIncrement],
- action: PolicyAction,
+ action: IpsecSpdAction.InputType,
inbound: bool,
sa_id: Optional[ObjIncrement] = None,
- proto: Optional[int] = None,
+ proto: IPsecProto.InputType = None,
laddr_range: Optional[NetworkIncrement] = None,
raddr_range: Optional[NetworkIncrement] = None,
lport_range: Optional[str] = None,
:param n_entries: Number of SPD entries to be added.
:param spd_id: SPD ID to add entries on.
:param priority: SPD entries priority, higher number = higher priority.
- :param action: Policy action.
+ :param action: IPsec SPD action.
:param inbound: If True policy is for inbound traffic, otherwise
outbound.
- :param sa_id: SAD entry ID for action PolicyAction.PROTECT.
+ :param sa_id: SAD entry ID for action IpsecSpdAction.PROTECT.
:param proto: Policy selector next layer protocol number.
:param laddr_range: Policy selector local IPv4 or IPv6 address range
in format IP/prefix or IP/mask. If no mask is provided,
:type n_entries: int
:type spd_id: int
:type priority: Optional[ObjIncrement]
- :type action: PolicyAction
+ :type action: IpsecSpdAction.InputType
:type inbound: bool
:type sa_id: Optional[ObjIncrement]
- :type proto: Optional[int]
+ :type proto: IPsecProto.InputType
:type laddr_range: Optional[NetworkIncrement]
:type raddr_range: Optional[NetworkIncrement]
:type lport_range: Optional[str]
:type rport_range: Optional[str]
:type is_ipv6: bool
"""
+ action = get_enum_instance(IpsecSpdAction, action)
+ proto = get_enum_instance(IPsecProto, proto)
if laddr_range is None:
laddr_range = "::/0" if is_ipv6 else "0.0.0.0/0"
laddr_range = NetworkIncrement(ip_network(laddr_range), 0)
if1_key: str,
if2_key: str,
n_tunnels: int,
- crypto_alg: CryptoAlg,
- integ_alg: Optional[IntegAlg],
+ crypto_alg: CryptoAlg.InputType,
+ integ_alg: IntegAlg.InputType,
raddr_ip2: Union[IPv4Address, IPv6Address],
addr_incr: int,
spi_d: dict,
:type if1_key: str
:type if2_key: str
:type n_tunnels: int
- :type crypto_alg: CryptoAlg
- :type integ_alg: Optional[IntegAlg]
+ :type crypto_alg: CryptoAlg.InputType
+ :type integ_alg: IntegAlg.InputType
:type raddr_ip2: Union[IPv4Address, IPv6Address]
:type addr_incr: int
:type spi_d: dict
:returns: Generated ckeys and ikeys.
:rtype: List[bytes], List[bytes]
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
if not existing_tunnels:
loop_sw_if_idx = IPsecUtil._ipsec_create_loopback_dut1_papi(
nodes, tun_ips, if1_key, if2_key
sad_entry = dict(
sad_id=None,
spi=None,
- protocol=int(IPsecProto.IPSEC_API_PROTO_ESP),
+ protocol=IPsecProto.ESP,
crypto_algorithm=crypto_alg.alg_int_repr,
crypto_key=c_key,
- integrity_algorithm=integ_alg.alg_int_repr if integ_alg else 0,
+ integrity_algorithm=integ_alg.alg_int_repr,
integrity_key=i_key,
flags=common_flags,
tunnel=dict(
)
args = dict(entry=sad_entry)
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))
- )
+ ckeys.append(gen_key(crypto_alg.key_len))
+ ikeys.append(gen_key(integ_alg.key_len))
# SAD entry for outband / tx path
sad_entry["sad_id"] = i
sad_entry["spi"] = spi_d["spi_1"] + i
tun_ips: dict,
if2_key: str,
n_tunnels: int,
- crypto_alg: CryptoAlg,
+ crypto_alg: CryptoAlg.InputType,
ckeys: Sequence[bytes],
- integ_alg: Optional[IntegAlg],
+ integ_alg: IntegAlg.InputType,
ikeys: Sequence[bytes],
raddr_ip1: Union[IPv4Address, IPv6Address],
addr_incr: int,
:type tun_ips: dict
:type if2_key: str
:type n_tunnels: int
- :type crypto_alg: CryptoAlg
+ :type crypto_alg: CryptoAlg.InputType
:type ckeys: Sequence[bytes]
- :type integ_alg: Optional[IntegAlg]
+ :type integ_alg: IntegAlg.InputType
:type ikeys: Sequence[bytes]
:type raddr_ip1: Union[IPv4Address, IPv6Address]
:type addr_incr: int
:type spi_d: dict
:type existing_tunnels: int
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
with PapiSocketExecutor(nodes["DUT2"], is_async=True) as papi_exec:
if not existing_tunnels:
# Set IP address on VPP node 2 interface
sad_entry = dict(
sad_id=None,
spi=None,
- protocol=int(IPsecProto.IPSEC_API_PROTO_ESP),
+ protocol=IPsecProto.ESP,
crypto_algorithm=crypto_alg.alg_int_repr,
crypto_key=c_key,
- integrity_algorithm=integ_alg.alg_int_repr if integ_alg else 0,
+ integrity_algorithm=integ_alg.alg_int_repr,
integrity_key=i_key,
flags=common_flags,
tunnel=dict(
)
args = dict(entry=sad_entry)
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))
- )
+ ckeys.append(gen_key(crypto_alg.key_len))
+ ikeys.append(gen_key(integ_alg.key_len))
# SAD entry for outband / tx path
sad_entry["sad_id"] = 100000 + i
sad_entry["spi"] = spi_d["spi_2"] + i
if1_key: str,
if2_key: str,
n_tunnels: int,
- crypto_alg: CryptoAlg,
- integ_alg: Optional[IntegAlg],
+ crypto_alg: CryptoAlg.InputType,
+ integ_alg: IntegAlg.InputType,
raddr_ip1: str,
raddr_ip2: str,
raddr_range: int,
:type if1_key: str
:type if2_key: str
:type n_tunnels: int
- :type crypto_alg: CryptoAlg
- :type integ_alg: Optional[IntegAlg]
+ :type crypto_alg: CryptoAlg.InputType
+ :type integ_alg: IntegAlg.InputType
:type raddr_ip1: str
:type raddr_ip2: str
:type raddr_range: int
:returns: Ckeys, ikeys, spi_1, spi_2.
:rtype: Optional[Tuple[List[bytes], List[bytes], int, int]]
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
n_tunnels = int(n_tunnels)
existing_tunnels = int(existing_tunnels)
spi_d = dict(spi_1=100000, spi_2=200000)
)
scp_node(nodes[dut], script_filename, script_filename)
- @staticmethod
- def vpp_ipsec_create_tunnel_interfaces_in_containers(
- nodes: dict,
- if1_ip_addr: str,
- if2_ip_addr: str,
- n_tunnels: int,
- crypto_alg: CryptoAlg,
- integ_alg: Optional[IntegAlg],
- raddr_ip1: str,
- raddr_ip2: str,
- raddr_range: int,
- n_instances: int,
- ) -> None:
- """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 n_tunnels: Number of tunnell interfaces to create.
- :param crypto_alg: The encryption algorithm name.
- :param integ_alg: The integrity algorithm name.
- :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.
- :param n_instances: Number of containers.
- :type nodes: dict
- :type if1_ip_addr: str
- :type if2_ip_addr: str
- :type n_tunnels: int
- :type crypto_alg: CryptoAlg
- :type integ_alg: Optional[IntegAlg]
- :type raddr_ip1: str
- :type raddr_ip2: str
- :type raddr_range: int
- :type n_instances: int
- """
- spi_1 = 100000
- spi_2 = 200000
- addr_incr = 1 << (32 - raddr_range)
-
- dut1_scripts = IPsecUtil._create_ipsec_script_files("DUT1", n_instances)
- dut2_scripts = IPsecUtil._create_ipsec_script_files("DUT2", n_instances)
-
- for cnf in range(0, n_instances):
- dut1_scripts[cnf].write(
- "create loopback interface\nset interface state loop0 up\n\n"
- )
- dut2_scripts[cnf].write(
- f"ip route add {if1_ip_addr}/8 via"
- f" {ip_address(if2_ip_addr) + cnf + 100} memif1/{cnf + 1}\n\n"
- )
-
- for tnl in range(0, n_tunnels):
- cnf = tnl % n_instances
- ckey = getattr(
- gen_key(IPsecUtil.get_crypto_alg_key_len(crypto_alg)), "hex"
- )
- integ = ""
- ikey = getattr(
- gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg)), "hex"
- )
- if integ_alg:
- 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(
- "set interface ip address loop0"
- f" {ip_address(if1_ip_addr) + tnl * addr_incr}/32\n"
- "create ipsec tunnel"
- 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 {tnl // n_instances}"
- f" salt 0x0 {integ}\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 * 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 * 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 {tnl // n_instances}"
- f" salt 0x0 {integ}\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(
- "DUT1", nodes, n_instances, dut1_scripts
- )
- IPsecUtil._close_and_copy_ipsec_script_files(
- "DUT2", nodes, n_instances, dut2_scripts
- )
-
@staticmethod
def vpp_ipsec_add_multiple_tunnels(
nodes: dict,
interface1: Union[str, int],
interface2: Union[str, int],
n_tunnels: int,
- crypto_alg: CryptoAlg,
- integ_alg: Optional[IntegAlg],
+ crypto_alg: CryptoAlg.InputType,
+ integ_alg: IntegAlg.InputType,
tunnel_ip1: str,
tunnel_ip2: str,
raddr_ip1: str,
:type interface1: Union[str, int]
:type interface2: Union[str, int]
:type n_tunnels: int
- :type crypto_alg: CryptoAlg
- :type integ_alg: Optional[IntegAlg]
+ :type crypto_alg: CryptoAlg.InputType
+ :type integ_alg: IntegAlg.InputType
:type tunnel_ip1: str
:type tunnel_ip2: str
:type raddr_ip1: str
:type raddr_range: int
:type tunnel_addr_incr: bool
"""
+ crypto_alg = get_enum_instance(CryptoAlg, crypto_alg)
+ integ_alg = get_enum_instance(IntegAlg, integ_alg)
+
spd_id = 1
p_hi = 100
p_lo = 10
spi_1 = 300000
spi_2 = 400000
- crypto_key = gen_key(
- IPsecUtil.get_crypto_alg_key_len(crypto_alg)
- ).decode()
- integ_key = (
- gen_key(IPsecUtil.get_integ_alg_key_len(integ_alg)).decode()
- if integ_alg
- else ""
- )
-
+ crypto_key = gen_key(crypto_alg.key_len).decode()
+ integ_key = gen_key(integ_alg.key_len).decode()
rmac = (
Topology.get_interface_mac(nodes["DUT2"], interface2)
if "DUT2" in nodes.keys()
nodes["DUT1"],
spd_id,
p_hi,
- PolicyAction.BYPASS,
+ IpsecSpdAction.BYPASS,
inbound=False,
- proto=50,
+ proto=IPsecProto.ESP,
laddr_range=dut1_local_outbound_range,
raddr_range=dut1_remote_outbound_range,
)
nodes["DUT1"],
spd_id,
p_hi,
- PolicyAction.BYPASS,
+ IpsecSpdAction.BYPASS,
inbound=True,
- proto=50,
+ proto=IPsecProto.ESP,
laddr_range=dut1_remote_outbound_range,
raddr_range=dut1_local_outbound_range,
)
n_tunnels,
spd_id,
priority=ObjIncrement(p_lo, 0),
- action=PolicyAction.PROTECT,
+ action=IpsecSpdAction.PROTECT,
inbound=False,
sa_id=ObjIncrement(sa_id_1, 1),
raddr_range=NetworkIncrement(ip_network(raddr_ip2)),
n_tunnels,
spd_id,
priority=ObjIncrement(p_lo, 0),
- action=PolicyAction.PROTECT,
+ action=IpsecSpdAction.PROTECT,
inbound=True,
sa_id=ObjIncrement(sa_id_2, 1),
raddr_range=NetworkIncrement(ip_network(raddr_ip1)),
nodes["DUT2"],
spd_id,
p_hi,
- PolicyAction.BYPASS,
+ IpsecSpdAction.BYPASS,
inbound=False,
- proto=50,
+ proto=IPsecProto.ESP,
laddr_range=dut2_remote_outbound_range,
raddr_range=dut2_local_outbound_range,
)
nodes["DUT2"],
spd_id,
p_hi,
- PolicyAction.BYPASS,
+ IpsecSpdAction.BYPASS,
inbound=True,
- proto=50,
+ proto=IPsecProto.ESP,
laddr_range=dut2_local_outbound_range,
raddr_range=dut2_remote_outbound_range,
)
n_tunnels,
spd_id,
priority=ObjIncrement(p_lo, 0),
- action=PolicyAction.PROTECT,
+ action=IpsecSpdAction.PROTECT,
inbound=True,
sa_id=ObjIncrement(sa_id_1, 1),
raddr_range=NetworkIncrement(ip_network(raddr_ip2)),
n_tunnels,
spd_id,
priority=ObjIncrement(p_lo, 0),
- action=PolicyAction.PROTECT,
+ action=IpsecSpdAction.PROTECT,
inbound=False,
sa_id=ObjIncrement(sa_id_2, 1),
raddr_range=NetworkIncrement(ip_network(raddr_ip1)),
@staticmethod
def vpp_ipsec_flow_enable_rss(
- node: dict, proto: str, rss_type: str, function: str = "default"
+ node: dict,
+ proto: str = "IPSEC_ESP",
+ rss_type: str = "esp",
+ function: str = "default",
) -> int:
"""Ipsec flow enable rss action.
:param proto: The flow protocol.
:param rss_type: RSS type.
:param function: RSS function.
-
:type node: dict
- :type proto: str
+ :type proto: IPsecProto.InputType
:type rss_type: str
:type function: str
:returns: flow_index.
:rtype: int
"""
+ # The proto argument does not correspond to IPsecProto.
+ # The allowed values come from src/vnet/ip/protocols.def
+ # and we do not have a good enum for that yet.
+ # FlowUti. and FlowUtil. are close but not exactly the same.
+
# TODO: to be fixed to use full PAPI when it is ready in VPP
cmd = (
f"test flow add src-ip any proto {proto} rss function"
Code Review / csit.git / blobdiff