path: root/resources/libraries/python/GBP.py

# Copyright (c) 2019 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.IPUtil import IPUtil
from resources.libraries.python.L2Util import L2Util
from resources.libraries.python.PapiExecutor import PapiSocketExecutor
from resources.libraries.python.topology import Topology


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
    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 GBP(object):
    """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 = 'gbp_route_domain_add'
        err_msg = 'Failed to add GBP route domain on {node}!'\
                  .format(node=node['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 = 'gbp_bridge_domain_add'
        err_msg = 'Failed to add GBP bridge domain on {node}!'\
                  .format(node=node['host'])

        args_in = dict(
            bd = dict(
                flags = getattr(GBPBridgeDomainFlags,
                                '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: dic
}
#!/usr/bin/env python3

import ipaddress
import unittest

from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto
from vpp_ip_route import VppIpRoute, VppRoutePath
from util import fragment_rfc791, fragment_rfc8200

import scapy.compat
from scapy.layers.l2 import Ether
from scapy.packet import Raw
from scapy.layers.inet import IP, UDP, ICMP, TCP
from scapy.layers.inet6 import IPv6, ICMPv6TimeExceeded, IPv6ExtHdrFragment, \
    ICMPv6EchoRequest, ICMPv6DestUnreach


class TestMAP(VppTestCase):
    """ MAP Test Case """

    @classmethod
    def setUpClass(cls):
        super(TestMAP, cls).setUpClass()

    @classmethod
    def tearDownClass(cls):
        super(TestMAP, cls).tearDownClass()

    def setUp(self):
        super(TestMAP, self).setUp()

        # create 2 pg interfaces
        self.create_pg_interfaces(range(4))

        # pg0 is 'inside' IPv4
        self.pg0.admin_up()
        self.pg0.config_ip4()
        self.pg0.resolve_arp()
        self.pg0.generate_remote_hosts(2)
        self.pg0.configure_ipv4_neighbors()

        # pg1 is 'outside' IPv6
        self.pg1.admin_up()
        self.pg1.config_ip6()
        self.pg1.generate_remote_hosts(4)
        self.pg1.configure_ipv6_neighbors()

    def tearDown(self):
        super(TestMAP, self).tearDown()
        for i in self.pg_interfaces:
            i.unconfig_ip4()
            i.unconfig_ip6()
            i.admin_down()

    def send_and_assert_encapped(self, packets, ip6_src, ip6_dst, dmac=None):
        if not dmac:
            dmac = self.pg1.remote_mac

        self.pg0.add_stream(packets)

        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        capture = self.pg1.get_capture(len(packets))
        for rx, tx in zip(capture, packets):
            self.assertEqual(rx[Ether].dst, dmac)
            self.assertEqual(rx[IP].src, tx[IP].src)
            self.assertEqual(rx[IPv6].src, ip6_src)
            self.assertEqual(rx[IPv6].dst, ip6_dst)

    def send_and_assert_encapped_one(self, packet, ip6_src, ip6_dst,
                                     dmac=None):
        return self.send_and_assert_encapped([packet], ip6_src, ip6_dst, dmac)

    def test_api_map_domain_dump(self):
        map_dst = '2001::/64'
        map_src = '3000::1/128'
        client_pfx = '192.168.0.0/16'
        tag = 'MAP-E tag.'
        index = self.vapi.map_add_domain(ip4_prefix=client_pfx,
                                         ip6_prefix=map_dst,
                                         ip6_src=map_src,
                                         tag=tag).index
        rv = self.vapi.map_domain_dump()

        # restore the state early so as to not impact subsequent tests.
        # If an assert fails, we will not get the chance to do it at the end.
        self.vapi.map_del_domain(index=index)

        self.assertGreater(len(rv), 0,
                           "Expected output from 'map_domain_dump'")

        # typedefs are returned as ipaddress objects.
        # wrap results in str() ugh! to avoid the need to call unicode.
        self.assertEqual(str(rv[0].ip4_prefix), client_pfx)
        self.assertEqual(str(rv[0].ip6_prefix), map_dst)
        self.assertEqual(str(rv[0].ip6_src), map_src)

        self.assertEqual(rv[0].tag, tag,
                         "output produced incorrect tag value.")

    def create_domains(self, ip4_pfx_str, ip6_pfx_str, ip6_src_str):
        ip4_pfx = ipaddress.ip_network(ip4_pfx_str)
        ip6_dst = ipaddress.ip_network(ip6_pfx_str)
        mod = ip4_pfx.num_addresses / 1024
        indicies = []
        for i in range(ip4_pfx.num_addresses):
            rv = self.vapi.map_add_domain(ip6_prefix=ip6_pfx_str,
                                          ip4_prefix=str(ip4_pfx[i]) + "/32",
                                          ip6_src=ip6_src_str)
            indicies.append(rv.index)
        return indicies

    def test_api_map_domains_get(self):
        # Create a bunch of domains
        no_domains = 4096  # This must be large enough to ensure VPP suspends
        domains = self.create_domains('130.67.0.0/20', '2001::/32',
                                      '2001::1/128')
        self.assertEqual(len(domains), no_domains)

        d = []
        cursor = 0

        # Invalid cursor
        rv, details = self.vapi.map_domains_get(cursor=no_domains+10)
        self.assertEqual(rv.retval, -7)

        # Delete a domain in the middle of walk
        rv, details = self.vapi.map_domains_get(cursor=0)
        self.assertEqual(rv.retval, -165)
        self.vapi.map_del_domain(index=rv.cursor)
        domains.remove(rv.cursor)

        # Continue at point of deleted cursor
        rv, details