X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Flibraries%2Fpython%2FSRv6.py;h=eca22a2b1744457bd9b4258a9996f38ab3d1cc31;hp=1099c754e63a0421100b068fd81ff7a5c8ce2a8b;hb=b6606e7625e308a66bdfb9d5a9c065b58e429a99;hpb=857bd6959f421ddc71003bd64d4337b8f0d0e60e diff --git a/resources/libraries/python/SRv6.py b/resources/libraries/python/SRv6.py index 1099c754e6..eca22a2b17 100644 --- a/resources/libraries/python/SRv6.py +++ b/resources/libraries/python/SRv6.py @@ -1,4 +1,4 @@ -# 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: @@ -14,10 +14,13 @@ """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 @@ -46,29 +49,15 @@ class SRv6Behavior(IntEnum): 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. @@ -76,12 +65,15 @@ class SRv6(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( @@ -123,105 +115,86 @@ class SRv6(object): :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) @@ -233,15 +206,11 @@ class SRv6(object): :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. @@ -253,32 +222,16 @@ class SRv6(object): :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) @@ -290,12 +243,8 @@ class SRv6(object): :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( @@ -316,40 +265,43 @@ class SRv6(object): :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): @@ -373,65 +325,22 @@ class SRv6(object): :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) @@ -443,12 +352,8 @@ class SRv6(object): :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): @@ -459,10 +364,12 @@ class SRv6(object): :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)