-# Copyright (c) 2019 Cisco and/or its affiliates.
+# Copyright (c) 2021 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:
"""Segment Routing over IPv6 data plane utilities library."""
from enum import IntEnum
+
from ipaddress import ip_address, IPv6Address
from resources.libraries.python.Constants import Constants
from resources.libraries.python.InterfaceUtil import InterfaceUtil
+from resources.libraries.python.IPAddress import IPAddress
+from resources.libraries.python.IPUtil import IPUtil
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
class SRv6PolicySteeringTypes(IntEnum):
- """SRv6 LocalSID supported functions."""
+ """SRv6 steering types."""
SR_STEER_L2 = 2
SR_STEER_IPV4 = 4
SR_STEER_IPV6 = 6
-class SRv6(object):
+class SRv6:
"""SRv6 class."""
- def __init__(self):
- pass
-
- @staticmethod
- def create_srv6_sid_object(ip_addr):
- """Create SRv6 SID object.
-
- :param ip_addr: IPv6 address.
- :type ip_addr: str
- :returns: SRv6 SID object.
- :rtype: dict
- """
- return dict(addr=IPv6Address(unicode(ip_addr)).packed)
-
@staticmethod
def create_srv6_sid_list(sids):
"""Create SRv6 SID list object.
:param sids: SID IPv6 addresses.
:type sids: list
:returns: SRv6 SID list object.
- :rtype: list
+ :rtype: dict
"""
- sid_list = list(0 for _ in xrange(16))
- for i in xrange(len(sids)):
- sid_list[i] = SRv6.create_srv6_sid_object(sids[i])
- return dict(num_sids=len(sids), weight=1, sids=sid_list)
+ sid_list = [IPv6Address(sid).packed for sid in sids]
+
+ return dict(
+ num_sids=len(sid_list),
+ weight=1,
+ sids=sid_list + (16 - len(sid_list)) * [IPv6Address(0).packed]
+ )
@staticmethod
def configure_sr_localsid(
:type sid_list: list
:raises ValueError: If required parameter is missing.
"""
- beh = behavior.replace('.', '_').upper()
+ beh = behavior.replace(u".", u"_").upper()
# There is no SRv6Behaviour enum defined for functions from SRv6 plugins
# so we need to use CLI command to configure it.
- if beh in (getattr(SRv6Behavior, 'END_AD').name,
- getattr(SRv6Behavior, 'END_AS').name,
- getattr(SRv6Behavior, 'END_AM').name):
- if beh == getattr(SRv6Behavior, 'END_AS').name:
+ if beh in (getattr(SRv6Behavior, u"END_AD").name,
+ getattr(SRv6Behavior, u"END_AS").name,
+ getattr(SRv6Behavior, u"END_AM").name):
+ if beh == getattr(SRv6Behavior, u"END_AS").name:
if next_hop is None or out_if is None or in_if is None or \
src_addr is None or sid_list is None:
raise ValueError(
- 'Required parameter(s) missing.\n'
- 'next_hop:{nh}\n'
- 'out_if:{oif}\n'
- 'in_if:{iif}\n'
- 'src_addr:{saddr}\n'
- 'sid_list:{sids}'.format(
- nh=next_hop, oif=out_if, iif=in_if, saddr=src_addr,
- sids=sid_list))
- sid_conf = 'next ' + ' next '.join(sid_list)
- params = 'nh {nh} oif {oif} iif {iif} src {saddr} {sids}'.\
- format(nh=next_hop, oif=out_if, iif=in_if, saddr=src_addr,
- sids=sid_conf)
+ f"Required parameter(s) missing.\n"
+ f"next_hop:{next_hop}\n "
+ f"out_if:{out_if}\n"
+ f"in_if:{in_if}\n"
+ f"src_addr:{src_addr}\n"
+ f"sid_list:{sid_list}"
+ )
+ sid_conf = f"next {u' next '.join(sid_list)}"
+ params = f"nh {next_hop} oif {out_if} iif {in_if} " \
+ f"src {src_addr} {sid_conf}"
else:
if next_hop is None or out_if is None or in_if is None:
raise ValueError(
- 'Required parameter(s) missing.\nnext_hop:{0}\n'
- 'out_if:{1}\nin_if:{2}'.format(next_hop, out_if, in_if))
- params = 'nh {0} oif {1} iif {2}'.format(
- next_hop, out_if, in_if)
+ f"Required parameter(s) missing.\n"
+ f"next_hop:{next_hop}\n"
+ f"out_if:{out_if}\n"
+ f"in_if:{in_if}"
+ )
+ params = f"nh {next_hop} oif {out_if} iif {in_if}"
- cli_cmd = 'sr localsid address {l_sid} behavior {beh} {params}'.\
- format(l_sid=local_sid, beh=behavior, params=params)
+ cli_cmd = f"sr localsid address {local_sid} behavior {behavior} " \
+ f"{params}"
PapiSocketExecutor.run_cli_cmd(node, cli_cmd)
return
- cmd = 'sr_localsid_add_del'
+ cmd = u"sr_localsid_add_del"
args = dict(
- is_del=0,
- localsid=SRv6.create_srv6_sid_object(local_sid),
- end_psp=0,
+ is_del=False,
+ localsid=IPv6Address(local_sid).packed,
+ end_psp=False,
behavior=getattr(SRv6Behavior, beh).value,
sw_if_index=Constants.BITWISE_NON_ZERO,
vlan_index=0,
fib_table=0,
- nh_addr6=0,
- nh_addr4=0
+ nh_addr=0
)
- err_msg = 'Failed to add SR localSID {lsid} on host {host}'.format(
- lsid=local_sid, host=node['host'])
-
- if beh in (getattr(SRv6Behavior, 'END_X').name,
- getattr(SRv6Behavior, 'END_DX4').name,
- getattr(SRv6Behavior, 'END_DX6').name):
+ err_msg = f"Failed to add SR localSID {local_sid} " \
+ f"host {node[u'host']}"
+ if beh in (getattr(SRv6Behavior, u"END_X").name,
+ getattr(SRv6Behavior, u"END_DX4").name,
+ getattr(SRv6Behavior, u"END_DX6").name):
if interface is None or next_hop is None:
- raise ValueError('Required parameter(s) missing.\n'
- 'interface:{ifc}\n'
- 'next_hop:{nh}'.
- format(ifc=interface, nh=next_hop))
- args['sw_if_index'] = InterfaceUtil.get_interface_index(
- node, interface)
- next_hop = ip_address(unicode(next_hop))
- if next_hop.version == 6:
- args['nh_addr6'] = next_hop.packed
- else:
- args['nh_addr4'] = next_hop.packed
- elif beh == getattr(SRv6Behavior, 'END_DX2').name:
+ raise ValueError(
+ f"Required parameter(s) missing.\n"
+ f"interface:{interface}\n"
+ f"next_hop:{next_hop}"
+ )
+ args[u"sw_if_index"] = InterfaceUtil.get_interface_index(
+ node, interface
+ )
+ args[u"nh_addr"] = IPAddress.create_ip_address_object(
+ ip_address(next_hop)
+ )
+ elif beh == getattr(SRv6Behavior, u"END_DX2").name:
if interface is None:
- raise ValueError('Required parameter missing.\ninterface:{ifc}'.
- format(ifc=interface))
- args['sw_if_index'] = InterfaceUtil.get_interface_index(
- node, interface)
- elif beh in (getattr(SRv6Behavior, 'END_DT4').name,
- getattr(SRv6Behavior, 'END_DT6').name):
+ raise ValueError(
+ f"Required parameter missing.\ninterface: {interface}"
+ )
+ args[u"sw_if_index"] = InterfaceUtil.get_interface_index(
+ node, interface
+ )
+ elif beh in (getattr(SRv6Behavior, u"END_DT4").name,
+ getattr(SRv6Behavior, u"END_DT6").name):
if fib_table is None:
- raise ValueError('Required parameter missing.\n'
- 'fib_table: {fib}'.format(fib=fib_table))
- args['fib_table'] = fib_table
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd, **args).get_reply(err_msg)
-
- @staticmethod
- def delete_sr_localsid(node, local_sid):
- """Delete SRv6 LocalSID on the given node.
-
- :param node: Given node to delete localSID on.
- :param local_sid: LocalSID IPv6 address.
- :type node: dict
- :type local_sid: str
- """
- cmd = 'sr_localsid_add_del'
- args = dict(
- is_del=1,
- localsid=SRv6.create_srv6_sid_object(local_sid),
- sw_if_index=Constants.BITWISE_NON_ZERO,
- )
- err_msg = 'Failed to delete SR localSID {lsid} on host {host}'.format(
- lsid=local_sid, host=node['host'])
+ raise ValueError(
+ f"Required parameter missing.\n"
+ f"fib_table: {fib_table}"
+ )
+ args[u"fib_table"] = fib_table
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args).get_reply(err_msg)
:param node: Given node to show localSIDs on.
:type node: dict
"""
- cmd = 'sr_localsids_dump'
- err_msg = 'Failed to get SR localSID dump on host {host}'.format(
- host=node['host'])
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd).get_details(err_msg)
+ cmd = u"sr_localsids_dump"
+ PapiSocketExecutor.dump_and_log(node, (cmd,))
@staticmethod
- def configure_sr_policy(node, bsid, sid_list, mode='encap'):
+ def configure_sr_policy(node, bsid, sid_list, mode=u"encap"):
"""Create SRv6 policy on the given node.
:param node: Given node to create SRv6 policy on.
:type sid_list: list
:type mode: str
"""
- cmd = 'sr_policy_add'
+ cmd = u"sr_policy_add"
args = dict(
- bsid_addr=IPv6Address(unicode(bsid)).packed,
+ bsid_addr=IPv6Address(bsid).packed,
weight=1,
- is_encap=1 if mode == 'encap' else 0,
+ is_encap=bool(mode == u"encap"),
+ is_spray=False,
sids=SRv6.create_srv6_sid_list(sid_list)
)
- err_msg = 'Failed to add SR policy for BindingSID {bsid} ' \
- 'on host {host}'.format(bsid=bsid, host=node['host'])
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd, **args).get_reply(err_msg)
-
- @staticmethod
- def delete_sr_policy(node, bsid):
- """Delete SRv6 policy on the given node.
-
- :param node: Given node to delete SRv6 policy on.
- :param bsid: BindingSID IPv6 address.
- :type node: dict
- :type bsid: str
- """
- cmd = 'sr_policy_del'
- args = dict(bsid_addr=IPv6Address(unicode(bsid)).packed)
- err_msg = 'Failed to delete SR policy for BindingSID {bsid} ' \
- 'on host {host}'.format(bsid=bsid, host=node['host'])
+ err_msg = f"Failed to add SR policy for BindingSID {bsid} " \
+ f"on host {node[u'host']}"
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args).get_reply(err_msg)
:param node: Given node to show SRv6 policies on.
:type node: dict
"""
- cmd = 'sr_policies_dump'
- err_msg = 'Failed to get SR policies dump on host {host}'.format(
- host=node['host'])
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd).get_details(err_msg)
+ cmd = u"sr_policies_dump"
+ PapiSocketExecutor.dump_and_log(node, (cmd,))
@staticmethod
def _get_sr_steer_policy_args(
:type interface: str
:type ip_addr: str
:type prefix: int
- :returns: Values for sw_if_index, mask_width, prefix_addr and
- traffic_type
+ :returns: Values for sw_if_index, prefix and traffic_type
:rtype: tuple
:raises ValueError: If unsupported mode used or required parameter
is missing.
"""
- if mode.lower() == 'l2':
+ if mode.lower() == u"l2":
if interface is None:
- raise ValueError('Required data missing.\ninterface:{ifc}'.
- format(ifc=interface))
+ raise ValueError(
+ f"Required data missing.\n"
+ f"interface: {interface}"
+ )
sw_if_index = InterfaceUtil.get_interface_index(node, interface)
- mask_width = 0
- prefix_addr = 16 * b'\x00'
- traffic_type = getattr(SRv6PolicySteeringTypes, 'SR_STEER_L2').value
- elif mode.lower() == 'l3':
+ prefix = 0
+ traffic_type = getattr(
+ SRv6PolicySteeringTypes, u"SR_STEER_L2"
+ ).value
+ elif mode.lower() == u"l3":
if ip_addr is None or prefix is None:
- raise ValueError('Required data missing.\nIP address:{0}\n'
- 'mask:{1}'.format(ip_addr, prefix))
+ raise ValueError(
+ f"Required data missing.\n"
+ f"IP address:{ip_addr}\n"
+ f"mask:{prefix}"
+ )
sw_if_index = Constants.BITWISE_NON_ZERO
- ip_addr = ip_address(unicode(ip_addr))
- prefix_addr = ip_addr.packed
- mask_width = int(prefix)
- if ip_addr.version == 4:
- prefix_addr += 12 * b'\x00'
- traffic_type = getattr(
- SRv6PolicySteeringTypes, 'SR_STEER_IPV4').value
- else:
- traffic_type = getattr(
- SRv6PolicySteeringTypes, 'SR_STEER_IPV6').value
+ ip_addr = ip_address(ip_addr)
+ prefix = IPUtil.create_prefix_object(ip_addr, int(prefix))
+ traffic_type = getattr(
+ SRv6PolicySteeringTypes, u"SR_STEER_IPV4"
+ ).value if ip_addr.version == 4 else getattr(
+ SRv6PolicySteeringTypes, u"SR_STEER_IPV6"
+ ).value
else:
- raise ValueError('Unsupported mode: {0}'.format(mode))
+ raise ValueError(f"Unsupported mode: {mode}")
- return sw_if_index, mask_width, prefix_addr, traffic_type
+ return sw_if_index, prefix, traffic_type
+ # TODO: Bring L1 names, arguments and defaults closer to PAPI ones.
@staticmethod
def configure_sr_steer(
node, mode, bsid, interface=None, ip_addr=None, prefix=None):
:raises ValueError: If unsupported mode used or required parameter
is missing.
"""
- sw_if_index, mask_width, prefix_addr, traffic_type = \
- SRv6._get_sr_steer_policy_args(
- node, mode, interface, ip_addr, prefix)
-
- cmd = 'sr_steering_add_del'
- args = dict(
- is_del=0,
- bsid_addr=IPv6Address(unicode(bsid)).packed,
- sr_policy_index=0,
- table_id=0,
- prefix_addr=prefix_addr,
- mask_width=mask_width,
- sw_if_index=sw_if_index,
- traffic_type=traffic_type
+ sw_if_index, prefix, traffic_type = SRv6._get_sr_steer_policy_args(
+ node, mode, interface, ip_addr, prefix
)
- err_msg = 'Failed to add SRv6 steering policy for BindingSID {bsid} ' \
- 'on host {host}'.format(bsid=bsid, host=node['host'])
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd, **args).get_reply(err_msg)
-
- @staticmethod
- def delete_sr_steer(
- node, mode, bsid, interface=None, ip_addr=None, mask=None):
- """Delete SRv6 steering policy on the given node.
-
- :param node: Given node to delete steering policy on.
- :param mode: Mode of operation - L2 or L3.
- :param bsid: BindingSID - local SID IPv6 address.
- :param interface: Interface name (Optional, required in case of
- L2 mode).
- :param ip_addr: IPv4/IPv6 address (Optional, required in case of L3
- mode).
- :param mask: IP address mask (Optional, required in case of L3 mode).
- :type node: dict
- :type mode: str
- :type bsid: str
- :type interface: str
- :type ip_addr: str
- :type mask: int
- :raises ValueError: If unsupported mode used or required parameter
- is missing.
- """
- sw_if_index, mask_width, prefix_addr, traffic_type = \
- SRv6._get_sr_steer_policy_args(node, mode, interface, ip_addr, mask)
- cmd = 'sr_steering_add_del'
+ cmd = u"sr_steering_add_del"
args = dict(
- is_del=1,
- bsid_addr=IPv6Address(unicode(bsid)).packed,
+ is_del=False,
+ bsid_addr=IPv6Address(str(bsid)).packed,
sr_policy_index=0,
table_id=0,
- prefix_addr=prefix_addr,
- mask_width=mask_width,
+ prefix=prefix,
sw_if_index=sw_if_index,
traffic_type=traffic_type
)
- err_msg = 'Failed to delete SRv6 steering policy for BindingSID ' \
- '{bsid} on host {host}'.format(bsid=bsid, host=node['host'])
+ err_msg = f"Failed to add SRv6 steering policy for BindingSID {bsid} " \
+ f"on host {node[u'host']}"
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args).get_reply(err_msg)
:param node: Given node to show SRv6 steering policies on.
:type node: dict
"""
- cmd = 'sr_steering_pol_dump'
- err_msg = 'Failed to get SR localSID dump on host {host}'.format(
- host=node['host'])
-
- with PapiSocketExecutor(node) as papi_exec:
- papi_exec.add(cmd).get_details(err_msg)
+ cmd = u"sr_steering_pol_dump"
+ PapiSocketExecutor.dump_and_log(node, (cmd,))
@staticmethod
def set_sr_encaps_source_address(node, ip6_addr):
:type node: dict
:type ip6_addr: str
"""
- cmd = 'sr_set_encap_source'
- args = dict(encaps_source=IPv6Address(unicode(ip6_addr)).packed)
- err_msg = 'Failed to set SRv6 encapsulation source address {addr} ' \
- 'on host {host}'.format(addr=ip6_addr, host=node['host'])
+ cmd = u"sr_set_encap_source"
+ args = dict(
+ encaps_source=IPv6Address(ip6_addr).packed
+ )
+ err_msg = f"Failed to set SRv6 encapsulation source address " \
+ f"{ip6_addr} on host {node[u'host']}"
with PapiSocketExecutor(node) as papi_exec:
papi_exec.add(cmd, **args).get_reply(err_msg)