diff options
Diffstat (limited to 'resources/libraries/python/GBP.py')
| -rw-r--r-- | resources/libraries/python/GBP.py | 346 |
1 files changed, 0 insertions, 346 deletions
diff --git a/resources/libraries/python/GBP.py b/resources/libraries/python/GBP.py deleted file mode 100644 index 9d56d4475b..0000000000 --- a/resources/libraries/python/GBP.py +++ /dev/null @@ -1,346 +0,0 @@ -# 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. - -"""GBP utilities library.""" - -from enum import IntEnum - -from ipaddress import ip_address - -from resources.libraries.python.IPAddress import IPAddress -from resources.libraries.python.L2Util import L2Util -from resources.libraries.python.PapiExecutor import PapiSocketExecutor - - -class GBPEndpointFlags(IntEnum): - """GBP Endpoint Flags.""" - GBP_API_ENDPOINT_FLAG_NONE = 0 - GBP_API_ENDPOINT_FLAG_BOUNCE = 1 - GBP_API_ENDPOINT_FLAG_REMOTE = 2 - GBP_API_ENDPOINT_FLAG_LEARNT = 4 - GBP_API_ENDPOINT_FLAG_EXTERNAL = 8 - - -class GBPBridgeDomainFlags(IntEnum): - """GBP Bridge Domain Flags.""" - GBP_BD_API_FLAG_NONE = 0 - GBP_BD_API_FLAG_DO_NOT_LEARN = 1 - GBP_BD_API_FLAG_UU_FWD_DROP = 2 - GBP_BD_API_FLAG_MCAST_DROP = 4 - GBP_BD_API_FLAG_UCAST_ARP = 8 - - -class GBPSubnetType(IntEnum): - """GBP Subnet Type.""" - GBP_API_SUBNET_TRANSPORT = 1 - # TODO: Names too long for pylint, fix in VPP. - GBP_API_SUBNET_STITCHED_INTERNAL = 2 - GBP_API_SUBNET_STITCHED_EXTERNAL = 3 - GBP_API_SUBNET_L3_OUT = 4 - GBP_API_SUBNET_ANON_L3_OUT = 5 - - -class GBPExtItfFlags(IntEnum): - """GBP External Interface Flags.""" - GBP_API_EXT_ITF_F_NONE = 0 - GBP_API_EXT_ITF_F_ANON = 1 - - -class GBPRuleAction(IntEnum): - """GBP Rule Action.""" - GBP_API_RULE_PERMIT = 1 - GBP_API_RULE_DENY = 2 - GBP_API_RULE_REDIRECT = 3 - - -class GBPHashMode(IntEnum): - """GBP Hash Mode.""" - GBP_API_HASH_MODE_SRC_IP = 1 - GBP_API_HASH_MODE_DST_IP = 2 - GBP_API_HASH_MODE_SYMETRIC = 3 - - -class GBP: - """GBP utilities.""" - - @staticmethod - def gbp_route_domain_add( - node, rd_id=1, ip4_table_id=1, ip6_table_id=0, - ip4_uu_sw_if_index=0xffffffff, ip6_uu_sw_if_index=0xffffffff): - """Add GBP route domain. - - :param node: Node to add GBP route domain on. - :param rd_id: GBP route domain ID. - :param ip4_table_id: IPv4 table. - :param ip6_table_id: IPv6 table. - :param ip4_uu_sw_if_index: IPv4 unicast interface index. - :param ip6_uu_sw_if_index: IPv6 unicast interface index. - :type node: dict - :type rd_id: int - :type ip4_table_id: int - :type ip6_table_id: int - :type ip4_uu_sw_if_index: int - :type ip6_uu_sw_if_index: int - """ - cmd = u"gbp_route_domain_add" - err_msg = f"Failed to add GBP route domain on {node[u'host']}!" - - args_in = dict( - rd=dict( - rd_id=rd_id, - ip4_table_id=ip4_table_id, - ip6_table_id=ip6_table_id, - ip4_uu_sw_if_index=ip4_uu_sw_if_index, - ip6_uu_sw_if_index=ip6_uu_sw_if_index - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_bridge_domain_add( - node, bvi_sw_if_index, bd_id=1, rd_id=1, - uu_fwd_sw_if_index=0xffffffff, bm_flood_sw_if_index=0xffffffff): - """Add GBP bridge domain. - - :param node: Node to add GBP bridge domain on. - :param bvi_sw_if_index: SW index of BVI/loopback interface. - :param bd_id: GBP bridge domain ID. - :param rd_id: GBP route domain ID. - :param uu_fwd_sw_if_index: Unicast forward interface index. - :param bm_flood_sw_if_index: Bcast/Mcast flood interface index. - :type node: dict - :type bvi_sw_if_index: int - :type bd_id: int - :type rd_id: int - :type uu_fwd_sw_if_index: int - :type bm_flood_sw_if_index: int - """ - cmd = u"gbp_bridge_domain_add" - err_msg = f"Failed to add GBP bridge domain on {node[u'host']}!" - - args_in = dict( - bd=dict( - flags=getattr( - GBPBridgeDomainFlags, u"GBP_BD_API_FLAG_NONE" - ).value, - bvi_sw_if_index=bvi_sw_if_index, - uu_fwd_sw_if_index=uu_fwd_sw_if_index, - bm_flood_sw_if_index=bm_flood_sw_if_index, - bd_id=bd_id, - rd_id=rd_id - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_endpoint_group_add( - node, sclass, bd_id=1, rd_id=1, vnid=1, - uplink_sw_if_index=0xffffffff, remote_ep_timeout=0xffffffff): - """Add GBP endpoint group. - - :param node: Node to add GBP endpoint group on. - :param sclass: Source CLASS. - :param bd_id: GBP bridge domain ID. - :param rd_id: GBP route domain ID. - :param uplink_sw_if_index: Uplink interface index. - :param remote_ep_timeout: Remote endpoint interface index. - :param vnid: VNID. - :type node: dict - :type sclass: int - :type bd_id: int - :type rd_id: int - :type vnid: int - :type uplink_sw_if_index: int - :type remote_ep_timeout: int - """ - cmd = u"gbp_endpoint_group_add" - err_msg = f"Failed to add GBP endpoint group on {node[u'host']}!" - - args_in = dict( - epg=dict( - uplink_sw_if_index=uplink_sw_if_index, - bd_id=bd_id, - rd_id=rd_id, - vnid=vnid, - sclass=sclass, - retention=dict( - remote_ep_timeout=remote_ep_timeout - ) - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_endpoint_add(node, sw_if_index, ip_addr, mac_addr, sclass): - """Add GBP endpoint. - - :param node: Node to add GBP endpoint on. - :param sw_if_index: SW index of interface. - :param ip_addr: GBP route domain ID. - :param mac_addr: MAC address. - :param sclass: Source CLASS. - :type node: dict - :type sw_if_index: int - :type ip_addr: str - :type mac_addr: str - :type sclass: int - """ - cmd = u"gbp_endpoint_add" - err_msg = f"Failed to add GBP endpoint on {node[u'host']}!" - - ips = list() - ips.append(IPAddress.create_ip_address_object(ip_address(ip_addr))) - tun_src = IPAddress.create_ip_address_object(ip_address(u"0.0.0.0")) - tun_dst = IPAddress.create_ip_address_object(ip_address(u"0.0.0.0")) - - args_in = dict( - endpoint=dict( - sw_if_index=sw_if_index, - ips=ips, - n_ips=len(ips), - mac=L2Util.mac_to_bin(mac_addr), - sclass=sclass, - flags=getattr( - GBPEndpointFlags, u"GBP_API_ENDPOINT_FLAG_EXTERNAL" - ).value, - tun=dict( - src=tun_src, - dst=tun_dst - ) - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_ext_itf_add_del(node, sw_if_index, bd_id=1, rd_id=1): - """Add external interface to GBP. - - :param node: Node to add external GBP interface on. - :param sw_if_index: SW index of interface. - :param bd_id: GBP bridge domain ID. - :param rd_id: GBP route domain ID. - :type node: dict - :type sw_if_index: int - :type bd_id: int - :type rd_id: int - """ - cmd = u"gbp_ext_itf_add_del" - err_msg = u"Failed to add external GBP interface on {node[u'host']}!" - - args_in = dict( - is_add=True, - ext_itf=dict( - sw_if_index=sw_if_index, - bd_id=bd_id, - rd_id=rd_id, - flags=getattr(GBPExtItfFlags, u"GBP_API_EXT_ITF_F_NONE").value - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_subnet_add_del( - node, address, subnet_length, sclass, rd_id=1, - sw_if_index=0xffffffff): - """Add external interface to GBP. - - :param node: Node to add GBP subnet on. - :param address: IPv4 adddress. - :param subnet_length: IPv4 address subnet. - :param sclass: Source CLASS. - :param rd_id: GBP route domain ID. - :param sw_if_index: Interface index. - :type node: dict - :type address: int - :type subnet_length: int - :type sclass: int - :type rd_id: int - :type sw_if_index: int - """ - cmd = u"gbp_subnet_add_del" - err_msg = f"Failed to add GBP subnet on {node[u'host']}!" - - args_in = dict( - is_add=True, - subnet=dict( - type=getattr(GBPSubnetType, u"GBP_API_SUBNET_L3_OUT").value, - sw_if_index=sw_if_index, - sclass=sclass, - prefix=dict( - address=IPAddress.create_ip_address_object( - ip_address(address) - ), - len=int(subnet_length) - ), - rd_id=rd_id - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) - - @staticmethod - def gbp_contract_add_del(node, sclass, dclass, acl_index=0, hash_mode=None): - """Add GBP contract. - - :param node: Node to add GBP contract on. - :param sclass: Source CLASS. - :param dclass: Destination CLASS. - :param acl_index: Index of ACL rule. - :param hash_mode: GBP contract hash mode. - :type node: dict - :type sclass: int - :type dclass: int - :type acl_index: int - :type hash_mode: str - """ - cmd = u"gbp_contract_add_del" - err_msg = f"Failed to add GBP contract on {node[u'host']}!" - - hash_mode = u"GBP_API_HASH_MODE_SRC_IP" if hash_mode is None \ - else hash_mode - rule_permit = dict( - action=getattr(GBPRuleAction, u"GBP_API_RULE_PERMIT").value, - nh_set=dict( - hash_mode=getattr(GBPHashMode, hash_mode).value, - n_nhs=8, - nhs=[dict()]*8, - ) - ) - rules = [rule_permit, rule_permit] - - args_in = dict( - is_add=True, - contract=dict( - acl_index=acl_index, - sclass=sclass, - dclass=dclass, - n_rules=len(rules), - rules=rules, - n_ether_types=16, - allowed_ethertypes=[0x800, 0x86dd] + [0]*14 - ) - ) - - with PapiSocketExecutor(node) as papi_exec: - papi_exec.add(cmd, **args_in).get_reply(err_msg) |