summaryrefslogtreecommitdiffstats
path: root/src/vnet/crypto/crypto.h
AgeCommit message (Expand)AuthorFilesLines
2020-09-03crypto: Add async crypto APIsNathan Skrzypczak1-1/+0
2020-09-03crypto: SW scheduler async crypto enginePiotrX Kleski1-10/+34
2020-09-01crypto: add chacha20-poly1305 algoArtem Glazychev1-2/+5
2020-04-30crypto: introduce async crypto infraFan Zhang1-10/+317
2020-02-11ipsec: add support for chained buffersFilip Tehlar1-10/+85
2019-11-12crypto: not use vec api with opt_data[VNET_CRYPTO_N_OP_IDS]Lijian Zhang1-1/+2
2019-07-24ipsec: GCM, Anti-replay and ESN fixessNeale Ranns1-2/+1
2019-06-25ipsec: return error if the engine backend has no handler for theNeale Ranns1-0/+1
2019-06-07ipsec: remove the set_key APINeale Ranns1-2/+0
2019-06-05IPSEC: some CLI fixesNeale Ranns1-1/+1
2019-04-30crypto: enforce per-alg crypto key lengthBenoît Ganne1-14/+16
2019-04-26crypto, ipsec: change GCM IV handlingDamjan Marion1-1/+1
2019-04-25crypto: AES GCM IV length is always 12Damjan Marion1-1/+1
2019-04-25crypto: improve key handlingDamjan Marion1-8/+43
2019-04-16IPSEC: support GCM in ESPNeale Ranns1-1/+1
2019-04-14crypto: add support for AES-CTR cipherfituldo1-1/+4
2019-04-11crypto: add performace test to unittest pluginDamjan Marion1-0/+1
2019-04-07crypto: add support for AEAD and AES-GCMDamjan Marion1-42/+74
2019-03-29ipsec: esp-decrypt reworkDamjan Marion1-11/+7
2019-03-28crypto: add vnet_crypto_op_init (...)Damjan Marion1-0/+10
2019-03-27ipsec: compress ipsec_sa_t so data used by dataplane code fits in cachelineDamjan Marion1-4/+10
2019-03-26crypto: add set crypto handler CLIFilip Tehlar1-0/+5
2019-03-26ipsec: esp-encrypt reworkDamjan Marion1-0/+1
2019-03-20crypto: add hmac truncate optionDamjan Marion1-1/+1
2019-03-20tests: implement crypto tests per RFC2202Filip Tehlar1-0/+1
2019-03-19crypto: introduce crypto infraDamjan Marion1-0/+163
1 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 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 
#!/usr/bin/env python3

import unittest

from framework import VppTestCase, VppTestRunner
from vpp_ip import DpoProto

from scapy.packet import Raw
from scapy.layers.l2 import Ether
from scapy.layers.inet import IP, UDP, TCP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import IPv6, IPerror6, ICMPv6DestUnreach

import struct

from ipaddress import ip_address, ip_network, \
    IPv4Address, IPv6Address, IPv4Network, IPv6Network

from vpp_object import VppObject
from vpp_papi import VppEnum

N_PKTS = 15


def find_cnat_translation(test, id):
    ts = test.vapi.cnat_translation_dump()
    for t in ts:
        if id == t.translation.id:
            return True
    return False


class Ep(object):
    """ CNat endpoint """

    def __init__(self, ip, port, l4p=TCP):
        self.ip = ip
        self.port = port
        self.l4p = l4p

    def encode(self):
        return {'addr': self.ip,
                'port': self.port}

    @property
    def isV6(self):
        return ":" in self.ip

    def __str__(self):
        return ("%s:%d" % (self.ip, self.port))


class EpTuple(object):
    """ CNat endpoint """

    def __init__(self, src, dst):
        self.src = src
        self.dst = dst

    def encode(self):
        return {'src_ep': self.src.encode(),
                'dst_ep': self.dst.encode()}

    def __str__(self):
        return ("%s->%s" % (self.src, self.dst))


class VppCNatTranslation(VppObject):

    def __init__(self, test, iproto, vip, paths):
        self._test = test
        self.vip = vip
        self.iproto = iproto
        self.paths = paths
        self.encoded_paths = []
        for path in self.paths:
            self.encoded_paths.append(path.encode())

    @property
    def vl4_proto(self):
        ip_proto = VppEnum.vl_api_ip_proto_t
        return {
            UDP: ip_proto.IP_API_PROTO_UDP,
            TCP: ip_proto.IP_API_PROTO_TCP,
        }[self.iproto]

    def delete(self):
        r = self._test.vapi.cnat_translation_del(id=self.id)

    def add_vpp_config(self):
        r = self._test.vapi.cnat_translation_update(
            {'vip': self.vip.encode(),
             'ip_proto': self.vl4_proto,
             'n_paths': len(self.paths),
             'paths': self.encoded_paths})
        self._test.registry.register(self, self._test.logger)
        self.id = r.id

    def modify_vpp_config(self, paths):
        self.paths = paths
        self.encoded_paths = []
        for path in self.paths:
            self.encoded_paths.append(path.encode())

        r = self._test.vapi.cnat_translation_update(
            {'vip': self.vip.encode(),
             'ip_proto': self.vl4_proto,
             'n_paths': len(self.paths),
             'paths': self.encoded_paths})
        self._test.registry.register(self, self._test.logger)

    def remove_vpp_config(self):
        self._test.vapi.cnat_translation_del(self.id)

    def query_vpp_config(self):
        return find_cnat_translation(self._test, self.id)

    def object_id(self):
        return ("cnat-translation-%s" % (self.vip))

    def get_stats(self):
        c = self._test.statistics.get_counter("/net/cnat-translation")
        return c[0][self.id]


class VppCNATSourceNat(VppObject):

    def __init__(self, test, address, exclude_subnets=[]):
        self._test = test
        self.address = address
        self.exclude_subnets = exclude_subnets

    def add_vpp_config(self):
        a = ip_address(self.address)
        if 4 == a.version:
            self._test.vapi.cnat_set_snat_addresses(snat_ip4=self.address)
        else:
            self._test.vapi.cnat_set_snat_addresses(snat_ip6=self.address)
        for subnet in self.exclude_subnets:
            self.cnat_exclude_subnet(subnet, True)

    def cnat_exclude_subnet(self, exclude_subnet, isAdd=True):
        add = 1 if isAdd else 0
        self._test.vapi.cnat_add_del_snat_prefix(
            prefix=exclude_subnet, is_add=add)

    def query_vpp_config(self):
        return False

    def remove_vpp_config(self):
        return False


class TestCNatTranslation(VppTestCase):
    """ CNat Translation """
    extra_vpp_punt_config = ["cnat", "{",
                             "session-db-buckets", "64",
                             "session-cleanup-timeout", "0.1",
                             "session-max-age", "1",
                             "tcp-max-age", "1",
                             "scanner", "off", "}"]

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

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

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

        self.create_pg_interfaces(range(3))

        for i in self.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.resolve_arp()
            i.config_ip6()
            i.resolve_ndp()

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

    def cnat_create_translation(self, vip, nbr):
        ip_v = "ip6" if vip.isV6 else "ip4"
        dep = Ep(getattr(self.pg1.remote_hosts[nbr], ip_v), 4000 + nbr)
        sep = Ep("::", 0) if vip.isV6 else Ep("0.0.0.0", 0)
        t1 = VppCNatTranslation(
            self, vip.l4p, vip,
            [EpTuple(sep, dep), EpTuple(sep, dep)])
        t1.add_vpp_config()
        return t1

    def cnat_test_translation(self, t1, nbr, sports, isV6=False):
        ip_v = "ip6" if isV6 else "ip4"
        ip_class = IPv6 if isV6 else IP
        vip = t1.vip

        #
        # Flows
        #
        for src in self.pg0.remote_hosts:
            for sport in sports:
                # from client to vip
                p1 = (Ether(dst=self.pg0.local_mac,
                            src=src.mac) /
                      ip_class(src=getattr(src, ip_v), dst=vip.ip) /
                      vip.l4p(sport=sport, dport=vip.port) /
                      Raw())

                self.vapi.cli("trace add pg-input 1")
                rxs = self.send_and_expect(self.pg0,
                                           p1 * N_PKTS,
                                           self.pg1)

                for rx in rxs:
                    self.assert_packet_checksums_valid(rx)
                    self.assertEqual(
                        rx[ip_class].dst,
                        getattr(self.pg1.remote_hosts[nbr], ip_v))
                    self.assertEqual(rx[vip.l4p].dport, 4000 + nbr)
                    self.assertEqual(
                        rx[ip_class].src,
                        getattr(src, ip_v))
                    self.assertEqual(rx[vip.l4p].sport, sport)

                # from vip to client
                p1 = (Ether(dst=self.pg1.local_mac,
                            src=self.pg1.remote_mac) /
                      ip_class(src=getattr(
                          self.pg1.remote_hosts[nbr],
                          ip_v),
                          dst=getattr(src, ip_v)) /
                      vip.l4p(sport=4000 + nbr, dport=sport) /
                      Raw())

                rxs = self.send_and_expect(self.pg1,
                                           p1 * N_PKTS,
                                           self.pg0)

                for rx in rxs:
                    self.assert_packet_checksums_valid(rx)
                    self.assertEqual(
                        rx[ip_class].dst,
                        getattr(src, ip_v))
                    self.assertEqual(rx[vip.l4p].dport, sport)
                    self.assertEqual(rx[ip_class].src, vip.ip)
                    self.assertEqual(rx[vip.l4p].sport, vip.port)

                #
                # packets to the VIP that do not match a
                # translation are dropped
                #
                p1 = (Ether(dst=self.pg0.local_mac,
                            src=src.mac) /
                      ip_class(src=getattr(src, ip_v), dst=vip.ip) /
                      vip.l4p(sport=sport, dport=6666) /
                      Raw())

                self.send_and_assert_no_replies(self.pg0,
                                                p1 * N_PKTS,
                                                self.pg1)

                #
                # packets from the VIP that do not match a
                # session are forwarded
                #
                p1 = (Ether(dst=self.pg1.local_mac,
                            src=self.pg1.remote_mac) /
                      ip_class(src=getattr(
                          self.pg1.remote_hosts[nbr],
                          ip_v),
                          dst=getattr(src, ip_v)) /
                      vip.l4p(sport=6666, dport=sport) /
                      Raw())

                rxs = self.send_and_expect(self.pg1,
                                           p1 * N_PKTS,
                                           self.pg0)

        self.assertEqual(t1.get_stats()['packets'],
                         N_PKTS *
                         len(sports) *
                         len(self.pg0.remote_hosts))

    def cnat_test_translation_update(self, t1, sports, isV6=False):
        ip_v = "ip6" if isV6 else "ip4"
        ip_class = IPv6 if isV6 else IP
        vip = t1.vip

        #
        # modify the translation to use a different backend
        #
        dep = Ep(getattr(self.pg2, 'remote_' + ip_v), 5000)
        sep = Ep("::", 0) if isV6 else Ep("0.0.0.0", 0)
        t1.modify_vpp_config([EpTuple(sep, dep)])

        #
        # existing flows follow the old path
        #
        for src in self.pg0.remote_hosts:
            for sport in sports:
                # from client to vip
                p1 = (Ether(dst=self.pg0.local_mac,
                            src=src.mac) /
                      ip_class(src=getattr(src, ip_v), dst=vip.ip) /
                      vip.l4p(sport=sport, dport=vip.port) /
                      Raw())

                rxs = self.send_and_expect(self.pg0,
                                           p1 * N_PKTS,
                                           self.pg1)

        #
        # new flows go to the new backend
        #
        for src in self.pg0.remote_hosts:
            p1 = (Ether(dst=self.pg0.local_mac,
                        src=src.mac) /
                  ip_class(src=getattr(src, ip_v), dst=vip.ip) /
                  vip.l4p(sport=9999, dport=vip.port) /
                  Raw())

            rxs = self.send_and_expect(self.pg0,
                                       p1 * N_PKTS,
                                       self.pg2)

    def cnat_translation(self, vips, isV6=False):
        """ CNat Translation """

        ip_class = IPv6 if isV6 else IP
        ip_v = "ip6" if isV6 else "ip4"
        sports = [1234, 1233]

        #
        # turn the scanner off whilst testing otherwise sessions
        # will time out
        #
        self.vapi.cli("test cnat scanner off")

        sessions = self.vapi.cnat_session_dump()

        trs = []
        for nbr, vip in enumerate(vips):
            trs.append(self.cnat_create_translation(vip, nbr))

        self.logger.info(self.vapi.cli("sh cnat client"))
        self.logger.info(self.vapi.cli("sh cnat translation"))

        #
        # translations
        #
        for nbr, vip in enumerate(vips):
            self.cnat_test_translation(trs[nbr], nbr, sports, isV6=isV6)
            self.cnat_test_translation_update(trs[nbr], sports, isV6=isV6)
            if isV6:
                self.logger.info(self.vapi.cli(
                    "sh ip6 fib %s" % self.pg0.remote_ip6))
            else:
                self.logger.info(self.vapi.cli(
                    "sh ip fib %s" % self.pg0.remote_ip4))
            self.logger.info(self.vapi.cli("sh cnat session verbose"))

        #
        # turn the scanner back on and wait until the sessions
        # all disapper
        #
        self.vapi.cli("test cnat scanner on")

        n_tries = 0
        sessions = self.vapi.cnat_session_dump()
        while (len(sessions) and n_tries < 100):
            n_tries += 1
            sessions = self.vapi.cnat_session_dump()
            self.sleep(2)
            print(self.vapi.cli("show cnat session verbose"))

        self.assertTrue(n_tries < 100)
        self.vapi.cli("test cnat scanner off")

        #
        # load some flows again and purge
        #
        for vip in vips:
            for src in self.pg0.remote_hosts:
                for sport in sports:
                    # from client to vip
                    p1 = (Ether(dst=self.pg0.local_mac,
                                src=src.mac) /
                          ip_class(src=getattr(src, ip_v), dst=vip.ip) /
                          vip.l4p(sport=sport, dport=vip.port) /
                          Raw())
                    self.send_and_expect(self.pg0,
                                         p1 * N_PKTS,
                                         self.pg2)

        for tr in trs:
            tr.delete()

        self.assertTrue(self.vapi.cnat_session_dump())
        self.vapi.cnat_session_purge()
        self.assertFalse(self.vapi.cnat_session_dump())

    def test_icmp(self):
        vips = [
            Ep("30.0.0.1", 5555),
            Ep("30.0.0.2", 5554),
            Ep("30.0.0.2", 5553, UDP),
            Ep("30::1", 6666),
            Ep("30::2", 5553, UDP),
        ]
        sport = 1234

        self.pg0.generate_remote_hosts(len(vips))
        self.pg0.configure_ipv6_neighbors()
        self.pg0.configure_ipv4_neighbors()

        self.pg1.generate_remote_hosts(len(vips))
        self.pg1.configure_ipv6_neighbors()
        self.pg1.configure_ipv4_neighbors()

        self.vapi.cli("test cnat scanner off")
        trs = []
        for nbr, vip in enumerate(vips):
            trs.append(self.cnat_create_translation(vip, nbr))

        self.logger.info(self.vapi.cli("sh cnat client"))
        self.logger.info(self.vapi.cli("sh cnat translation"))

        for nbr, vip in enumerate(vips):
            if vip.isV6:
                client_addr = self.pg0.remote_hosts[0].ip6
                remote_addr = self.pg1.remote_hosts[nbr].ip6
                remote2_addr = self.pg2.remote_hosts[0].ip6
            else:
                client_addr = self.pg0.remote_hosts[0].ip4
                remote_addr = self.pg1.remote_hosts[nbr].ip4
                remote2_addr = self.pg2.remote_hosts[0].ip4
            IP46 = IPv6 if vip.isV6 else IP
            # from client to vip
            p1 = (Ether(dst=self.pg0.local_mac,
                        src=self.pg0.remote_hosts[0].mac) /
                  IP46(src=client_addr, dst=vip.ip) /
                  vip.l4p(sport=sport, dport=vip.port) /
                  Raw())

            rxs = self.send_and_expect(self.pg0,
                                       p1 * N_PKTS,
                                       self.pg1)

            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].dst, remote_addr)
                self.assertEqual(rx[vip.l4p].dport, 4000 + nbr)
                self.assertEqual(rx[IP46].src, client_addr)
                self.assertEqual(rx[vip.l4p].sport, sport)

            InnerIP = rxs[0][IP46]

            ICMP46 = ICMPv6DestUnreach if vip.isV6 else ICMP
            ICMPelem = ICMPv6DestUnreach(code=1) if vip.isV6 else ICMP(type=11)
            # from vip to client, ICMP error
            p1 = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
                  IP46(src=remote_addr, dst=client_addr) /
                  ICMPelem / InnerIP)

            rxs = self.send_and_expect(self.pg1,
                                       p1 * N_PKTS,
                                       self.pg0)

            TCPUDPError = TCPerror if vip.l4p == TCP else UDPerror
            IP46error = IPerror6 if vip.isV6 else IPerror
            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].src, vip.ip)
                self.assertEqual(rx[ICMP46][IP46error].src, client_addr)
                self.assertEqual(rx[ICMP46][IP46error].dst, vip.ip)
                self.assertEqual(rx[ICMP46][IP46error]
                                 [TCPUDPError].sport, sport)
                self.assertEqual(rx[ICMP46][IP46error]
                                 [TCPUDPError].dport, vip.port)

            # from other remote to client, ICMP error
            # outside shouldn't be NAT-ed
            p1 = (Ether(dst=self.pg2.local_mac, src=self.pg2.remote_mac) /
                  IP46(src=remote2_addr, dst=client_addr) /
                  ICMPelem / InnerIP)

            rxs = self.send_and_expect(self.pg1,
                                       p1 * N_PKTS,
                                       self.pg0)

            TCPUDPError = TCPerror if vip.l4p == TCP else UDPerror
            IP46error = IPerror6 if vip.isV6 else IPerror
            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].src, remote2_addr)
                self.assertEqual(rx[ICMP46][IP46error].src, client_addr)
                self.assertEqual(rx[ICMP46][IP46error].dst, vip.ip)
                self.assertEqual(rx[ICMP46][IP46error]
                                 [TCPUDPError].sport, sport)
                self.assertEqual(rx[ICMP46][IP46error]
                                 [TCPUDPError].dport, vip.port)

        self.vapi.cnat_session_purge()

    def test_cnat6(self):
        # """ CNat Translation ipv6 """
        vips = [
            Ep("30::1", 5555),
            Ep("30::2", 5554),
            Ep("30::2", 5553, UDP),
        ]

        self.pg0.generate_remote_hosts(len(vips))
        self.pg0.configure_ipv6_neighbors()
        self.pg1.generate_remote_hosts(len(vips))
        self.pg1.configure_ipv6_neighbors()

        self.cnat_translation(vips, isV6=True)

    def test_cnat4(self):
        # """ CNat Translation ipv4 """

        vips = [
            Ep("30.0.0.1", 5555),
            Ep("30.0.0.2", 5554),
            Ep("30.0.0.2", 5553, UDP),
        ]

        self.pg0.generate_remote_hosts(len(vips))
        self.pg0.configure_ipv4_neighbors()
        self.pg1.generate_remote_hosts(len(vips))
        self.pg1.configure_ipv4_neighbors()

        self.cnat_translation(vips)


class TestCNatSourceNAT(VppTestCase):
    """ CNat Source NAT """
    extra_vpp_punt_config = ["cnat", "{",
                             "session-max-age", "1",
                             "tcp-max-age", "1", "}"]

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

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

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

        self.create_pg_interfaces(range(3))

        for i in self.pg_interfaces:
            i.admin_up()
            i.config_ip4()
            i.resolve_arp()
            i.config_ip6()
            i.resolve_ndp()

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

    def cnat_set_snat_address(self, srcNatAddr, interface, isV6=False):
        t1 = VppCNATSourceNat(self, srcNatAddr)
        t1.add_vpp_config()
        cnat_arc_name = "ip6-unicast" if isV6 else "ip4-unicast"
        cnat_feature_name = "ip6-cnat-snat" if isV6 else "ip4-cnat-snat"
        self.vapi.feature_enable_disable(
            enable=1,
            arc_name=cnat_arc_name,
            feature_name=cnat_feature_name,
            sw_if_index=interface.sw_if_index)

        return t1

    def cnat_test_sourcenat(self, srcNatAddr, l4p=TCP, isV6=False):
        ip_v = "ip6" if isV6 else "ip4"
        IP46 = IPv6 if isV6 else IP
        sports = [1234, 1235, 1236]
        dports = [6661, 6662, 6663]

        self.pg0.generate_remote_hosts(1)
        self.pg0.configure_ipv4_neighbors()
        self.pg0.configure_ipv6_neighbors()
        self.pg1.generate_remote_hosts(len(sports))
        self.pg1.configure_ipv4_neighbors()
        self.pg1.configure_ipv6_neighbors()

        self.vapi.cli("test cnat scanner on")
        t1 = self.cnat_set_snat_address(srcNatAddr, self.pg0, isV6)

        for nbr, remote_host in enumerate(self.pg1.remote_hosts):
            if isV6:
                client_addr = self.pg0.remote_hosts[0].ip6
                remote_addr = self.pg1.remote_hosts[nbr].ip6
            else:
                client_addr = self.pg0.remote_hosts[0].ip4
                remote_addr = self.pg1.remote_hosts[nbr].ip4
            # from pods to outside network
            p1 = (
                Ether(dst=self.pg0.local_mac,
                      src=self.pg0.remote_hosts[0].mac) /
                IP46(src=client_addr, dst=remote_addr) /
                l4p(sport=sports[nbr], dport=dports[nbr]) /
                Raw())

            rxs = self.send_and_expect(
                self.pg0,
                p1 * N_PKTS,
                self.pg1)
            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].dst, remote_addr)
                self.assertEqual(rx[l4p].dport, dports[nbr])
                self.assertEqual(rx[IP46].src, srcNatAddr)
                sport = rx[l4p].sport

            # from outside to pods
            p2 = (
                Ether(dst=self.pg1.local_mac,
                      src=self.pg1.remote_hosts[nbr].mac) /
                IP46(src=remote_addr, dst=srcNatAddr) /
                l4p(sport=dports[nbr], dport=sport) /
                Raw())

            rxs = self.send_and_expect(
                self.pg1,
                p2 * N_PKTS,
                self.pg0)

            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].dst, client_addr)
                self.assertEqual(rx[l4p].dport, sports[nbr])
                self.assertEqual(rx[l4p].sport, dports[nbr])
                self.assertEqual(rx[IP46].src, remote_addr)

            # add remote host to exclude list
            subnet_mask = 100 if isV6 else 16
            subnet = "%s/%d" % (remote_addr, subnet_mask)
            exclude_subnet = ip_network(subnet, strict=False)

            t1.cnat_exclude_subnet(exclude_subnet)
            self.vapi.cnat_session_purge()

            rxs = self.send_and_expect(
                self.pg0,
                p1 * N_PKTS,
                self.pg1)
            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].dst, remote_addr)
                self.assertEqual(rx[l4p].dport, dports[nbr])
                self.assertEqual(rx[IP46].src, client_addr)

            # remove remote host from exclude list
            t1.cnat_exclude_subnet(exclude_subnet, isAdd=False)
            self.vapi.cnat_session_purge()

            rxs = self.send_and_expect(
                self.pg0,
                p1 * N_PKTS,
                self.pg1)

            for rx in rxs:
                self.assert_packet_checksums_valid(rx)
                self.assertEqual(rx[IP46].dst, remote_addr)
                self.assertEqual(rx[l4p].dport, dports[nbr])
                self.assertEqual(rx[IP46].src, srcNatAddr)

    def test_cnat6_sourcenat(self):
        # """ CNat Source Nat ipv6 """
        self.cnat_test_sourcenat(self.pg2.remote_hosts[0].ip6, TCP, True)
        self.cnat_test_sourcenat(self.pg2.remote_hosts[0].ip6, UDP, True)

    def test_cnat4_sourcenat(self):
        # """ CNat Source Nat ipv4 """
        self.cnat_test_sourcenat(self.pg2.remote_hosts[0].ip4, TCP)
        self.cnat_test_sourcenat(self.pg2.remote_hosts[0].ip4, UDP)


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