summaryrefslogtreecommitdiffstats
path: root/test/vpp_neighbor.py
AgeCommit message (Collapse)AuthorFilesLines
2022-05-10tests: replace pycodestyle with blackKlement Sekera1-23/+23
Drop pycodestyle for code style checking in favor of black. Black is much faster, stable PEP8 compliant code style checker offering also automatic formatting. It aims to be very stable and produce smallest diffs. It's used by many small and big projects. Running checkstyle with black takes a few seconds with a terse output. Thus, test-checkstyle-diff is no longer necessary. Expand scope of checkstyle to all python files in the repo, replacing test-checkstyle with checkstyle-python. Also, fixstyle-python is now available for automatic style formatting. Note: python virtualenv has been consolidated in test/Makefile, test/requirements*.txt which will eventually be moved to a central location. This is required to simply the automated generation of docker executor images in the CI. Type: improvement Change-Id: I022a326603485f58585e879ac0f697fceefbc9c8 Signed-off-by: Klement Sekera <klement.sekera@gmail.com> Signed-off-by: Dave Wallace <dwallacelf@gmail.com>
2021-03-25stats: python vpp_stats rewrite to access stat segment directlyOle Troan1-1/+1
This module implement Python access to the VPP statistics segment. It accesses the data structures directly in shared memory. VPP uses optimistic locking, so data structures may change underneath us while we are reading. Data is copied out and it's important to spend as little time as possible "holding the lock". Counters are stored in VPP as a two dimensional array. Index by thread and index (typically sw_if_index). Simple counters count only packets, Combined counters count packets and octets. Counters can be accessed in either dimension. stat['/if/rx'] - returns 2D lists stat['/if/rx'][0] - returns counters for all interfaces for thread 0 stat['/if/rx'][0][1] - returns counter for interface 1 on thread 0 stat['/if/rx'][0][1]['packets'] - returns the packet counter for interface 1 on thread 0 stat['/if/rx'][:, 1] - returns the counters for interface 1 on all threads stat['/if/rx'][:, 1].packets() - returns the packet counters for interface 1 on all threads stat['/if/rx'][:, 1].sum_packets() - returns the sum of packet counters for interface 1 on all threads stat['/if/rx-miss'][:, 1].sum() - returns the sum of packet counters for interface 1 on all threads for simple counters Type: refactor Signed-off-by: Ole Troan <ot@cisco.com> Change-Id: I1fe7f7c7d11378d06be8276db5e1900ecdb8f515 Signed-off-by: Ole Troan <ot@cisco.com>
2020-10-08ip-neighbor: Grat ARPs from different subnet are droppedNeale Ranns1-1/+2
Type: test Signed-off-by: Neale Ranns <nranns@cisco.com> Change-Id: Id07e8981a903f11f50fb494a93d01815382025e4
2020-04-28tests: implement ipaddress convenience methodsPaul Vinciguerra1-5/+2
Add vpp specific properties to ip addresses for use in the api. .vapi_af -- returns [ADDRESS_IP4, ADDRESS_IP6] .vapi_af_name -- returns the string ['ip4', 'ip6'] Update tests to demonstrate usage. Type: feature Change-Id: I43447a1522769d99f89debdc714c51700068d771 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
2019-12-17ip: Protocol Independent IP NeighborsNeale Ranns1-2/+5
Type: feature - ip-neighbour: generic neighbour handling; APIs, DBs, event handling, aging - arp: ARP protocol implementation - ip6-nd; IPv6 neighbor discovery implementation; separate ND, MLD, RA - ip6-link; manage link-local addresses - l2-arp-term; events separated from IP neighbours, since they are not the same. vnet retains just enough education to perform ND/ARP packet construction. arp and ip6-nd to be moved to plugins soon. Change-Id: I88dedd0006b299344f4c7024a0aa5baa6b9a8bbe Signed-off-by: Neale Ranns <nranns@cisco.com>
2019-11-05ip: Fragmentation fixesNeale Ranns1-0/+1
Type: fix if the packet is about to be fragmented, then don't call any of the actions that expect the rewrite to have been written. 1) don't double count packets thru the adjacency (original & fragments) 2) don't double decrement the TTL for fragments 3) return to ip4-midchain post ip-frag if that's where we started. 4) only run midchain/mcast fixups if not fragmenting (if no errors) Change-Id: Ib2866787a42713ee5871b87b597d8f74b901044b Signed-off-by: Neale Ranns <nranns@cisco.com>
2019-03-29tests: refactor vpp_object.pyPaul Vinciguerra1-3/+0
Move __str__ to super for all subclasses of VppObject Implement __repr__ in VppObject Implement __hash__ and __eq__ Change-Id: Ibd4ea37b84b17f499ab86630fb5b9ed9c8b4b1c2 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
2019-03-01Tests: Trivial fox use of 'is'.Paul Vinciguerra1-1/+1
'is' evaluates diffently depending on the value. >>> x=-10 >>> x is -10 False >>> x=10 >>> x is 10 True >>> Change-Id: If72ec7c47e3e95180a4d08a773984253c80154d6 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
2019-03-01Tests: Remove all wildcard imports.Paul Vinciguerra1-1/+1
They mask dependencies and confuse third party tools. Change-Id: I24ee16bfa0d746a8e8cb99891ba8b20523048b69 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
2019-02-26VPP-1508 python tests: unicodePaul Vinciguerra1-1/+5
Change unicode references to use text_type Change-Id: Ia71c16e3235bc509abd3b1c651ae125f892ab108 Signed-off-by: Paul Vinciguerra <pvinci@vinciconsulting.com>
2019-01-30Use IP and MAC API types for neighborsNeale Ranns1-27/+27
use address_t and mac_address_t for IPv6 and ARP entries and all other API calls in ip.api aprat from the route ones, that will follow in a separate commit Change-Id: I67161737c2184d3f8fc1e79ebd2b55121c5b0191 Signed-off-by: Neale Ranns <nranns@cisco.com>
2018-12-18PAPI: Add MACAddress object wrapper for vl_api_mac_address_tOle Troan1-3/+3
Change the definition of vl_api_mac_address_t to an aliased type. Change-Id: I1434f316d0fad6a099592f39bceeb8faeaf1d134 Signed-off-by: Ole Troan <ot@cisco.com>
2018-12-10Test framework: StringIO fixes for Python3Ole Troan1-1/+1
Add 2/3 support to binarytomac and mactobinary and move to vpp_mac.py Change-Id: I3dc7e4a24486aee22140c781aae7e44e58935877 Signed-off-by: Ole Troan <ot@cisco.com>
2018-10-01Add adjacency counters to the stats segmentNeale Ranns1-1/+6
Change-Id: I6b59df939c9daf40e261d73d19f500bd90abe6ff Signed-off-by: Neale Ranns <neale.ranns@cisco.com>
2018-06-08Gratuitous ARP packet handlingNeale Ranns1-2/+6
only learn from a GARP packet if it is an update to an existing entry. Change-Id: I4c1b59cfedb911466e5e4c9756cf53a6676e1909 Signed-off-by: Neale Ranns <neale.ranns@cisco.com>
2018-03-28test: fix VppNeighbor.query_vpp_configNeale Ranns1-14/+10
Change-Id: I2a879ee9d5065a14eb351eccd0350693eb1ff521 Signed-off-by: Neale Ranns <neale.ranns@cisco.com>
2017-04-19Learn IP6 link-local ND entries from NSs sourced from link-local addressNeale Ranns1-1/+1
Change-Id: I4c3ce4d58df7977490fc94991291422ea1e31ee3 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-03-09IMplementation for option to not create a FIB table entry when adding a ↵Neale Ranns1-2/+4
neighbor entry Change-Id: I952039e101031ee6a06e63f4c73d8eb359423e1a Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-03-03IPv6 RA improvementsNeale Ranns1-1/+1
1) tests for RA options 2) memleaks deleteing a ip6_radv_info_t 3) MLD prefix code refactoring Change-Id: I34db103994bd8fbdbbec50b202d72770dd145681 Signed-off-by: Neale Ranns <nranns@cisco.com>
2017-02-27[Proxy] ARP testsNeale Ranns1-0/+77
Change-Id: I40d6d763b55a26cdee0afef85d1acdd19dd10dd6 Signed-off-by: Neale Ranns <nranns@cisco.com>
54' href='#n654'>654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 
#!/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 test_map_e_udp(self):
        """ MAP-E UDP"""

        #
        # Add a route to the MAP-BR
        #
        map_br_pfx = "2001::"
        map_br_pfx_len = 32
        map_route = VppIpRoute(self,
                               map_br_pfx,
                               map_br_pfx_len,
                               [VppRoutePath(self.pg1.remote_ip6,
                                             self.pg1.sw_if_index)])
        map_route.add_vpp_config()

        #
        # Add a domain that maps from pg0 to pg1
        #
        map_dst = '2001::/32'
        map_src = '3000::1/128'
        client_pfx = '192.168.0.0/16'
        map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
        tag = 'MAP-E tag.'
        self.vapi.map_add_domain(ip4_prefix=client_pfx,
                                 ip6_prefix=map_dst,
                                 ip6_src=map_src,
                                 ea_bits_len=20,
                                 psid_offset=4,
                                 psid_length=4,
                                 tag=tag)

        self.vapi.map_param_set_security_check(enable=1, fragments=1)

        # Enable MAP on interface.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=0)

        # Ensure MAP doesn't steal all packets!
        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
              IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 100))
        rx = self.send_and_expect(self.pg0, v4 * 4, self.pg0)
        v4_reply = v4[1]
        v4_reply.ttl -= 1
        for p in rx:
            self.validate(p[1], v4_reply)

        #
        # Fire in a v4 packet that will be encapped to the BR
        #
        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
              IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 100))

        self.send_and_assert_encapped(v4 * 4, "3000::1", map_translated_addr)

        #
        # Verify reordered fragments are able to pass as well
        #
        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
              IP(id=1, src=self.pg0.remote_ip4, dst='192.168.1.1') /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 1000))

        frags = fragment_rfc791(v4, 400)
        frags.reverse()

        self.send_and_assert_encapped(frags, "3000::1", map_translated_addr)

        # Enable MAP on interface.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg1.sw_if_index,
                                        is_translation=0)

        # Ensure MAP doesn't steal all packets
        v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
              IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 100))
        rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
        v6_reply = v6[1]
        v6_reply.hlim -= 1
        for p in rx:
            self.validate(p[1], v6_reply)

        #
        # Fire in a V6 encapped packet.
        # expect a decapped packet on the inside ip4 link
        #
        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IPv6(dst='3000::1', src=map_translated_addr) /
             IP(dst=self.pg0.remote_ip4, src='192.168.1.1') /
             UDP(sport=10000, dport=20000) /
             Raw(b'\xa5' * 100))

        self.pg1.add_stream(p)

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

        rx = self.pg0.get_capture(1)
        rx = rx[0]

        self.assertFalse(rx.haslayer(IPv6))
        self.assertEqual(rx[IP].src, p[IP].src)
        self.assertEqual(rx[IP].dst, p[IP].dst)

        #
        # Verify encapped reordered fragments pass as well
        #
        p = (IP(id=1, dst=self.pg0.remote_ip4, src='192.168.1.1') /
             UDP(sport=10000, dport=20000) /
             Raw(b'\xa5' * 1500))
        frags = fragment_rfc791(p, 400)
        frags.reverse()

        stream = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
                  IPv6(dst='3000::1', src=map_translated_addr) /
                  x for x in frags)

        self.pg1.add_stream(stream)

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

        rx = self.pg0.get_capture(len(frags))

        for r in rx:
            self.assertFalse(r.haslayer(IPv6))
            self.assertEqual(r[IP].src, p[IP].src)
            self.assertEqual(r[IP].dst, p[IP].dst)

        # Verify that fragments pass even if ipv6 layer is fragmented
        stream = (IPv6(dst='3000::1', src=map_translated_addr) / x
                  for x in frags)

        v6_stream = [
            Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) / x
            for i in range(len(frags))
            for x in fragment_rfc8200(
                IPv6(dst='3000::1', src=map_translated_addr) / frags[i],
                i, 200)]

        self.pg1.add_stream(v6_stream)

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

        rx = self.pg0.get_capture(len(frags))

        for r in rx:
            self.assertFalse(r.haslayer(IPv6))
            self.assertEqual(r[IP].src, p[IP].src)
            self.assertEqual(r[IP].dst, p[IP].dst)

        #
        # Pre-resolve. No API for this!!
        #
        self.vapi.ppcli("map params pre-resolve ip6-nh 4001::1")

        self.send_and_assert_no_replies(self.pg0, v4,
                                        "resolved via default route")

        #
        # Add a route to 4001::1. Expect the encapped traffic to be
        # sent via that routes next-hop
        #
        pre_res_route = VppIpRoute(self, "4001::1", 128,
                                   [VppRoutePath(self.pg1.remote_hosts[2].ip6,
                                                 self.pg1.sw_if_index)])
        pre_res_route.add_vpp_config()

        self.send_and_assert_encapped_one(v4, "3000::1",
                                          map_translated_addr,
                                          dmac=self.pg1.remote_hosts[2].mac)

        #
        # change the route to the pre-solved next-hop
        #
        pre_res_route.modify([VppRoutePath(self.pg1.remote_hosts[3].ip6,
                                           self.pg1.sw_if_index)])
        pre_res_route.add_vpp_config()

        self.send_and_assert_encapped_one(v4, "3000::1",
                                          map_translated_addr,
                                          dmac=self.pg1.remote_hosts[3].mac)

        #
        # cleanup. The test infra's object registry will ensure
        # the route is really gone and thus that the unresolve worked.
        #
        pre_res_route.remove_vpp_config()
        self.vapi.ppcli("map params pre-resolve del ip6-nh 4001::1")

    def test_map_e_inner_frag(self):
        """ MAP-E Inner fragmentation """

        #
        # Add a route to the MAP-BR
        #
        map_br_pfx = "2001::"
        map_br_pfx_len = 32
        map_route = VppIpRoute(self,
                               map_br_pfx,
                               map_br_pfx_len,
                               [VppRoutePath(self.pg1.remote_ip6,
                                             self.pg1.sw_if_index)])
        map_route.add_vpp_config()

        #
        # Add a domain that maps from pg0 to pg1
        #
        map_dst = '2001::/32'
        map_src = '3000::1/128'
        client_pfx = '192.168.0.0/16'
        map_translated_addr = '2001:0:101:7000:0:c0a8:101:7'
        tag = 'MAP-E tag.'
        self.vapi.map_add_domain(ip4_prefix=client_pfx,
                                 ip6_prefix=map_dst,
                                 ip6_src=map_src,
                                 ea_bits_len=20,
                                 psid_offset=4,
                                 psid_length=4,
                                 mtu=1000,
                                 tag=tag)

        # Enable MAP on interface.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=0)

        # Enable inner fragmentation
        self.vapi.map_param_set_fragmentation(inner=1)

        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
              IP(src=self.pg0.remote_ip4, dst='192.168.1.1') /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 1300))

        self.pg_send(self.pg0, v4*1)
        rx = self.pg1.get_capture(2)

        frags = fragment_rfc791(v4[1], 1000)
        frags[0].id = 0
        frags[1].id = 0
        frags[0].ttl -= 1
        frags[1].ttl -= 1
        frags[0].chksum = 0
        frags[1].chksum = 0

        v6_reply1 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
                     frags[0])
        v6_reply2 = (IPv6(src='3000::1', dst=map_translated_addr, hlim=63) /
                     frags[1])
        rx[0][1].fl = 0
        rx[1][1].fl = 0
        rx[0][1][IP].id = 0
        rx[1][1][IP].id = 0
        rx[0][1][IP].chksum = 0
        rx[1][1][IP].chksum = 0

        self.validate(rx[0][1], v6_reply1)
        self.validate(rx[1][1], v6_reply2)

    def test_map_e_tcp_mss(self):
        """ MAP-E TCP MSS"""

        #
        # Add a route to the MAP-BR
        #
        map_br_pfx = "2001::"
        map_br_pfx_len = 32
        map_route = VppIpRoute(self,
                               map_br_pfx,
                               map_br_pfx_len,
                               [VppRoutePath(self.pg1.remote_ip6,
                                             self.pg1.sw_if_index)])
        map_route.add_vpp_config()

        #
        # Add a domain that maps from pg0 to pg1
        #
        map_dst = '2001::/32'
        map_src = '3000::1/128'
        client_pfx = '192.168.0.0/16'
        map_translated_addr = '2001:0:101:5000:0:c0a8:101:5'
        tag = 'MAP-E TCP tag.'
        self.vapi.map_add_domain(ip4_prefix=client_pfx,
                                 ip6_prefix=map_dst,
                                 ip6_src=map_src,
                                 ea_bits_len=20,
                                 psid_offset=4,
                                 psid_length=4,
                                 tag=tag)

        # Enable MAP on pg0 interface.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=0)

        # Enable MAP on pg1 interface.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg1.sw_if_index,
                                        is_translation=0)

        # TCP MSS clamping
        mss_clamp = 1300
        self.vapi.map_param_set_tcp(mss_clamp)

        #
        # Send a v4 packet that will be encapped.
        #
        p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.1.1')
        p_tcp = TCP(sport=20000, dport=30000, flags="S",
                    options=[("MSS", 1455)])
        p4 = p_ether / p_ip4 / p_tcp

        self.pg1.add_stream(p4)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()

        rx = self.pg1.get_capture(1)
        rx = rx[0]

        self.assertTrue(rx.haslayer(IPv6))
        self.assertEqual(rx[IP].src, p4[IP].src)
        self.assertEqual(rx[IP].dst, p4[IP].dst)
        self.assertEqual(rx[IPv6].src, "3000::1")
        self.assertEqual(rx[TCP].options,
                         TCP(options=[('MSS', mss_clamp)]).options)

    def validate(self, rx, expected):
        self.assertEqual(rx, expected.__class__(scapy.compat.raw(expected)))

    def validate_frag6(self, p6_frag, p_ip6_expected):
        self.assertFalse(p6_frag.haslayer(IP))
        self.assertTrue(p6_frag.haslayer(IPv6))
        self.assertTrue(p6_frag.haslayer(IPv6ExtHdrFragment))
        self.assertEqual(p6_frag[IPv6].src, p_ip6_expected.src)
        self.assertEqual(p6_frag[IPv6].dst, p_ip6_expected.dst)

    def validate_frag_payload_len6(self, rx, proto, payload_len_expected):
        payload_total = 0
        for p in rx:
            payload_total += p[IPv6].plen

        # First fragment has proto
        payload_total -= len(proto())

        # Every fragment has IPv6 fragment header
        payload_total -= len(IPv6ExtHdrFragment()) * len(rx)

        self.assertEqual(payload_total, payload_len_expected)

    def validate_frag4(self, p4_frag, p_ip4_expected):
        self.assertFalse(p4_frag.haslayer(IPv6))
        self.assertTrue(p4_frag.haslayer(IP))
        self.assertTrue(p4_frag[IP].frag != 0 or p4_frag[IP].flags.MF)
        self.assertEqual(p4_frag[IP].src, p_ip4_expected.src)
        self.assertEqual(p4_frag[IP].dst, p_ip4_expected.dst)

    def validate_frag_payload_len4(self, rx, proto, payload_len_expected):
        payload_total = 0
        for p in rx:
            payload_total += len(p[IP].payload)

        # First fragment has proto
        payload_total -= len(proto())

        self.assertEqual(payload_total, payload_len_expected)

    def payload(self, len):
        return 'x' * len

    def test_map_t(self):
        """ MAP-T """

        #
        # Add a domain that maps from pg0 to pg1
        #
        map_dst = '2001:db8::/32'
        map_src = '1234:5678:90ab:cdef::/64'
        ip4_pfx = '192.168.0.0/24'
        tag = 'MAP-T Tag.'

        self.vapi.map_add_domain(ip6_prefix=map_dst,
                                 ip4_prefix=ip4_pfx,
                                 ip6_src=map_src,
                                 ea_bits_len=16,
                                 psid_offset=6,
                                 psid_length=4,
                                 mtu=1500,
                                 tag=tag)

        # Enable MAP-T on interfaces.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=1)
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg1.sw_if_index,
                                        is_translation=1)

        # Ensure MAP doesn't steal all packets!
        v4 = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
              IP(src=self.pg0.remote_ip4, dst=self.pg0.remote_ip4) /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 100))
        rx = self.send_and_expect(self.pg0, v4*1, self.pg0)
        v4_reply = v4[1]
        v4_reply.ttl -= 1
        for p in rx:
            self.validate(p[1], v4_reply)
        # Ensure MAP doesn't steal all packets
        v6 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
              IPv6(src=self.pg1.remote_ip6, dst=self.pg1.remote_ip6) /
              UDP(sport=20000, dport=10000) /
              Raw(b'\xa5' * 100))
        rx = self.send_and_expect(self.pg1, v6*1, self.pg1)
        v6_reply = v6[1]
        v6_reply.hlim -= 1
        for p in rx:
            self.validate(p[1], v6_reply)

        map_route = VppIpRoute(self,
                               "2001:db8::",
                               32,
                               [VppRoutePath(self.pg1.remote_ip6,
                                             self.pg1.sw_if_index,
                                             proto=DpoProto.DPO_PROTO_IP6)])
        map_route.add_vpp_config()

        #
        # Send a v4 packet that will be translated
        #
        p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
        payload = TCP(sport=0xabcd, dport=0xabcd)

        p4 = (p_ether / p_ip4 / payload)
        p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
                              dst="2001:db8:1f0::c0a8:1:f") / payload)
        p6_translated.hlim -= 1
        rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
        for p in rx:
            self.validate(p[1], p6_translated)

        # Send back an IPv6 packet that will be "untranslated"
        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
        p6 = (p_ether6 / p_ip6 / payload)
        p4_translated = (IP(src='192.168.0.1',
                            dst=self.pg0.remote_ip4) / payload)
        p4_translated.id = 0
        p4_translated.ttl -= 1
        rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
        for p in rx:
            self.validate(p[1], p4_translated)

        # IPv4 TTL=0
        ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=0)
        p4 = (p_ether / ip4_ttl_expired / payload)

        icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
                          dst=self.pg0.remote_ip4) /
                       ICMP(type='time-exceeded',
                            code='ttl-zero-during-transit') /
                       IP(src=self.pg0.remote_ip4,
                          dst='192.168.0.1', ttl=0) / payload)
        rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
        for p in rx:
            self.validate(p[1], icmp4_reply)

        # IPv4 TTL=1
        ip4_ttl_expired = IP(src=self.pg0.remote_ip4, dst='192.168.0.1', ttl=1)
        p4 = (p_ether / ip4_ttl_expired / payload)

        icmp4_reply = (IP(id=0, ttl=254, src=self.pg0.local_ip4,
                          dst=self.pg0.remote_ip4) /
                       ICMP(type='time-exceeded',
                            code='ttl-zero-during-transit') /
                       IP(src=self.pg0.remote_ip4,
                          dst='192.168.0.1', ttl=1) / payload)
        rx = self.send_and_expect(self.pg0, p4*1, self.pg0)
        for p in rx:
            self.validate(p[1], icmp4_reply)

        # IPv6 Hop limit
        ip6_hlim_expired = IPv6(hlim=0, src='2001:db8:1ab::c0a8:1:ab',
                                dst='1234:5678:90ab:cdef:ac:1001:200:0')
        p6 = (p_ether6 / ip6_hlim_expired / payload)

        icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
                            dst="2001:db8:1ab::c0a8:1:ab") /
                       ICMPv6TimeExceeded(code=0) /
                       IPv6(src="2001:db8:1ab::c0a8:1:ab",
                            dst='1234:5678:90ab:cdef:ac:1001:200:0',
                            hlim=0) / payload)
        rx = self.send_and_expect(self.pg1, p6*1, self.pg1)
        for p in rx:
            self.validate(p[1], icmp6_reply)

        # IPv4 Well-known port
        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
        payload = UDP(sport=200, dport=200)
        p4 = (p_ether / p_ip4 / payload)
        self.send_and_assert_no_replies(self.pg0, p4*1)

        # IPv6 Well-known port
        payload = UDP(sport=200, dport=200)
        p6 = (p_ether6 / p_ip6 / payload)
        self.send_and_assert_no_replies(self.pg1, p6*1)

        # UDP packet fragmentation
        payload_len = 1453
        payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
        p4 = (p_ether / p_ip4 / payload)
        self.pg_enable_capture()
        self.pg0.add_stream(p4)
        self.pg_start()
        rx = self.pg1.get_capture(2)

        p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
                                dst='2001:db8:1e0::c0a8:1:e')
        for p in rx:
            self.validate_frag6(p, p_ip6_translated)

        self.validate_frag_payload_len6(rx, UDP, payload_len)

        # UDP packet fragmentation send fragments
        payload_len = 1453
        payload = UDP(sport=40000, dport=4000) / self.payload(payload_len)
        p4 = (p_ether / p_ip4 / payload)
        frags = fragment_rfc791(p4, fragsize=1000)
        self.pg_enable_capture()
        self.pg0.add_stream(frags)
        self.pg_start()
        rx = self.pg1.get_capture(2)

        for p in rx:
            self.validate_frag6(p, p_ip6_translated)

        self.validate_frag_payload_len6(rx, UDP, payload_len)

        # Send back an fragmented IPv6 UDP packet that will be "untranslated"
        payload = UDP(sport=4000, dport=40000) / self.payload(payload_len)
        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6 = IPv6(src='2001:db8:1e0::c0a8:1:e',
                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
        p6 = (p_ether6 / p_ip6 / payload)
        frags6 = fragment_rfc8200(p6, identification=0xdcba, fragsize=1000)

        p_ip4_translated = IP(src='192.168.0.1', dst=self.pg0.remote_ip4)
        p4_translated = (p_ip4_translated / payload)
        p4_translated.id = 0
        p4_translated.ttl -= 1

        self.pg_enable_capture()
        self.pg1.add_stream(frags6)
        self.pg_start()
        rx = self.pg0.get_capture(2)

        for p in rx:
            self.validate_frag4(p, p4_translated)

        self.validate_frag_payload_len4(rx, UDP, payload_len)

        # ICMP packet fragmentation
        payload = ICMP(id=6529) / self.payload(payload_len)
        p4 = (p_ether / p_ip4 / payload)
        self.pg_enable_capture()
        self.pg0.add_stream(p4)
        self.pg_start()
        rx = self.pg1.get_capture(2)

        p_ip6_translated = IPv6(src='1234:5678:90ab:cdef:ac:1001:200:0',
                                dst='2001:db8:160::c0a8:1:6')
        for p in rx:
            self.validate_frag6(p, p_ip6_translated)

        self.validate_frag_payload_len6(rx, ICMPv6EchoRequest, payload_len)

        # ICMP packet fragmentation send fragments
        payload = ICMP(id=6529) / self.payload(payload_len)
        p4 = (p_ether / p_ip4 / payload)
        frags = fragment_rfc791(p4, fragsize=1000)
        self.pg_enable_capture()
        self.pg0.add_stream(frags)
        self.pg_start()
        rx = self.pg1.get_capture(2)

        for p in rx:
            self.validate_frag6(p, p_ip6_translated)

        self.validate_frag_payload_len6(rx, ICMPv6EchoRequest, payload_len)

        # TCP MSS clamping
        self.vapi.map_param_set_tcp(1300)

        #
        # Send a v4 TCP SYN packet that will be translated and MSS clamped
        #
        p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
        payload = TCP(sport=0xabcd, dport=0xabcd, flags="S",
                      options=[('MSS', 1460)])

        p4 = (p_ether / p_ip4 / payload)
        p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
                              dst="2001:db8:1f0::c0a8:1:f") / payload)
        p6_translated.hlim -= 1
        p6_translated[TCP].options = [('MSS', 1300)]
        rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
        for p in rx:
            self.validate(p[1], p6_translated)

        # Send back an IPv6 packet that will be "untranslated"
        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
        p6 = (p_ether6 / p_ip6 / payload)
        p4_translated = (IP(src='192.168.0.1',
                            dst=self.pg0.remote_ip4) / payload)
        p4_translated.id = 0
        p4_translated.ttl -= 1
        p4_translated[TCP].options = [('MSS', 1300)]
        rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
        for p in rx:
            self.validate(p[1], p4_translated)

        # TCP MSS clamping cleanup
        self.vapi.map_param_set_tcp(0)

        # Enable icmp6 param to get back ICMPv6 unreachable messages in case
        # of security check fails
        self.vapi.map_param_set_icmp6(enable_unreachable=1)

        # Send back an IPv6 packet that will be droppped due to security
        # check fail
        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6_sec_check_fail = IPv6(src='2001:db8:1fe::c0a8:1:f',
                                    dst='1234:5678:90ab:cdef:ac:1001:200:0')
        payload = TCP(sport=0xabcd, dport=0xabcd)
        p6 = (p_ether6 / p_ip6_sec_check_fail / payload)

        self.pg_send(self.pg1, p6*1)
        self.pg0.get_capture(0, timeout=1)
        rx = self.pg1.get_capture(1)

        icmp6_reply = (IPv6(hlim=255, src=self.pg1.local_ip6,
                            dst='2001:db8:1fe::c0a8:1:f') /
                       ICMPv6DestUnreach(code=5) /
                       p_ip6_sec_check_fail / payload)

        for p in rx:
            self.validate(p[1], icmp6_reply)

        # ICMPv6 unreachable messages cleanup
        self.vapi.map_param_set_icmp6(enable_unreachable=0)

    def test_map_t_ip6_psid(self):
        """ MAP-T v6->v4 PSID validation"""

        #
        # Add a domain that maps from pg0 to pg1
        #
        map_dst = '2001:db8::/32'
        map_src = '1234:5678:90ab:cdef::/64'
        ip4_pfx = '192.168.0.0/24'
        tag = 'MAP-T Test Domain'

        self.vapi.map_add_domain(ip6_prefix=map_dst,
                                 ip4_prefix=ip4_pfx,
                                 ip6_src=map_src,
                                 ea_bits_len=16,
                                 psid_offset=6,
                                 psid_length=4,
                                 mtu=1500,
                                 tag=tag)

        # Enable MAP-T on interfaces.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=1)
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg1.sw_if_index,
                                        is_translation=1)

        map_route = VppIpRoute(self,
                               "2001:db8::",
                               32,
                               [VppRoutePath(self.pg1.remote_ip6,
                                             self.pg1.sw_if_index,
                                             proto=DpoProto.DPO_PROTO_IP6)])
        map_route.add_vpp_config()

        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
                     dst='1234:5678:90ab:cdef:ac:1001:200:0')

        # Send good IPv6 source port, ensure translated IPv4 received
        payload = TCP(sport=0xabcd, dport=80)
        p6 = (p_ether6 / p_ip6 / payload)
        p4_translated = (IP(src='192.168.0.1',
                            dst=self.pg0.remote_ip4) / payload)
        p4_translated.id = 0
        p4_translated.ttl -= 1
        rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
        for p in rx:
            self.validate(p[1], p4_translated)

        # Send bad IPv6 source port, ensure translated IPv4 not received
        payload = TCP(sport=0xdcba, dport=80)
        p6 = (p_ether6 / p_ip6 / payload)
        self.send_and_assert_no_replies(self.pg1, p6*1)

    def test_map_t_pre_resolve(self):
        """ MAP-T pre-resolve"""

        # Add a domain that maps from pg0 to pg1
        map_dst = '2001:db8::/32'
        map_src = '1234:5678:90ab:cdef::/64'
        ip4_pfx = '192.168.0.0/24'
        tag = 'MAP-T Test Domain.'

        self.vapi.map_add_domain(ip6_prefix=map_dst,
                                 ip4_prefix=ip4_pfx,
                                 ip6_src=map_src,
                                 ea_bits_len=16,
                                 psid_offset=6,
                                 psid_length=4,
                                 mtu=1500,
                                 tag=tag)

        # Enable MAP-T on interfaces.
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg0.sw_if_index,
                                        is_translation=1)
        self.vapi.map_if_enable_disable(is_enable=1,
                                        sw_if_index=self.pg1.sw_if_index,
                                        is_translation=1)

        # Enable pre-resolve option
        self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
                                                ip6_nh_address="4001::1",
                                                is_add=1)

        # Add a route to 4001::1 and expect the translated traffic to be
        # sent via that route next-hop.
        pre_res_route6 = VppIpRoute(self, "4001::1", 128,
                                    [VppRoutePath(self.pg1.remote_hosts[2].ip6,
                                                  self.pg1.sw_if_index)])
        pre_res_route6.add_vpp_config()

        # Add a route to 10.1.2.3 and expect the "untranslated" traffic to be
        # sent via that route next-hop.
        pre_res_route4 = VppIpRoute(self, "10.1.2.3", 32,
                                    [VppRoutePath(self.pg0.remote_hosts[1].ip4,
                                                  self.pg0.sw_if_index)])
        pre_res_route4.add_vpp_config()

        # Send an IPv4 packet that will be translated
        p_ether = Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac)
        p_ip4 = IP(src=self.pg0.remote_ip4, dst='192.168.0.1')
        payload = TCP(sport=0xabcd, dport=0xabcd)
        p4 = (p_ether / p_ip4 / payload)

        p6_translated = (IPv6(src="1234:5678:90ab:cdef:ac:1001:200:0",
                              dst="2001:db8:1f0::c0a8:1:f") / payload)
        p6_translated.hlim -= 1

        rx = self.send_and_expect(self.pg0, p4*1, self.pg1)
        for p in rx:
            self.assertEqual(p[Ether].dst, self.pg1.remote_hosts[2].mac)
            self.validate(p[1], p6_translated)

        # Send back an IPv6 packet that will be "untranslated"
        p_ether6 = Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac)
        p_ip6 = IPv6(src='2001:db8:1f0::c0a8:1:f',
                     dst='1234:5678:90ab:cdef:ac:1001:200:0')
        p6 = (p_ether6 / p_ip6 / payload)

        p4_translated = (IP(src='192.168.0.1',
                            dst=self.pg0.remote_ip4) / payload)
        p4_translated.id = 0
        p4_translated.ttl -= 1

        rx = self.send_and_expect(self.pg1, p6*1, self.pg0)
        for p in rx:
            self.assertEqual(p[Ether].dst, self.pg0.remote_hosts[1].mac)
            self.validate(p[1], p4_translated)

        # Cleanup pre-resolve option
        self.vapi.map_param_add_del_pre_resolve(ip4_nh_address="10.1.2.3",
                                                ip6_nh_address="4001::1",
                                                is_add=0)

if __name__ == '__main__':
    unittest.main(testRunner=VppTestRunner)