summaryrefslogtreecommitdiffstats
path: root/src/vnet/plugin/p1.c
diff options
context:
space:
mode:
authorDamjan Marion <damarion@cisco.com>2017-01-10 17:07:38 +0100
committerDave Barach <openvpp@barachs.net>2017-01-11 12:57:31 +0000
commit1c5ee872b14c5e9412b8c0e4f169b85d11238d14 (patch)
tree14afec369eeabaef32aabd38c667275effdba341 /src/vnet/plugin/p1.c
parente546d3b0f739d35ec2e4702181d99ff4190e8b46 (diff)
Remove vcgn plugin
Change-Id: I79f18ec386dedd91a8dcea2ca5726208b7b3c67c Signed-off-by: Damjan Marion <damarion@cisco.com>
Diffstat (limited to 'src/vnet/plugin/p1.c')
0 files changed, 0 insertions, 0 deletions
#n143'>143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179
#!/usr/bin/env python3

import ipaddress
import random
import socket
import struct
import unittest
from io import BytesIO
from time import sleep

import scapy.compat
from framework import VppTestCase, VppTestRunner, running_extended_tests
from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder
from scapy.all import bind_layers, Packet, ByteEnumField, ShortField, \
    IPField, IntField, LongField, XByteField, FlagsField, FieldLenField, \
    PacketListField
from scapy.data import IP_PROTOS
from scapy.layers.inet import IP, TCP, UDP, ICMP
from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror
from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6, IPv6ExtHdrFragment
from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply, \
    ICMPv6ND_NS, ICMPv6ND_NA, ICMPv6NDOptDstLLAddr, fragment6
from scapy.layers.l2 import Ether, ARP, GRE
from scapy.packet import Raw
from syslog_rfc5424_parser import SyslogMessage, ParseError
from syslog_rfc5424_parser.constants import SyslogSeverity
from util import ip4_range
from util import ppc, ppp
from vpp_acl import AclRule, VppAcl, VppAclInterface
from vpp_ip_route import VppIpRoute, VppRoutePath
from vpp_neighbor import VppNeighbor
from vpp_papi import VppEnum


class TestNAT66(VppTestCase):
    """ NAT66 Test Cases """

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

        cls.nat_addr = 'fd01:ff::2'
        cls.create_pg_interfaces(range(2))
        cls.interfaces = list(cls.pg_interfaces)

        for i in cls.interfaces:
            i.admin_up()
            i.config_ip6()
            i.configure_ipv6_neighbors()

    @property
    def config_flags(self):
        return VppEnum.vl_api_nat_config_flags_t

    def plugin_enable(self):
        self.vapi.nat66_plugin_enable_disable(enable=1)

    def plugin_disable(self):
        self.vapi.nat66_plugin_enable_disable(enable=0)

    def setUp(self):
        super(TestNAT66, self).setUp()
        self.plugin_enable()

    def tearDown(self):
        super(TestNAT66, self).tearDown()
        if not self.vpp_dead:
            self.plugin_disable()

    def test_static(self):
        """ 1:1 NAT66 test """
        flags = self.config_flags.NAT_IS_INSIDE
        self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
                                          sw_if_index=self.pg0.sw_if_index)
        self.vapi.nat66_add_del_interface(is_add=1,
                                          sw_if_index=self.pg1.sw_if_index)
        self.vapi.nat66_add_del_static_mapping(
            local_ip_address=self.pg0.remote_ip6,
            external_ip_address=self.nat_addr,
            is_add=1)

        # in2out
        pkts = []
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
             TCP())
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
             UDP())
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
             ICMPv6EchoRequest())
        pkts.append(p)
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
             GRE() / IP() / TCP())
        pkts.append(p)
        self.pg0.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(len(pkts))

        for packet in capture:
            try:
                self.assertEqual(packet[IPv6].src, self.nat_addr)
                self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
                self.assert_packet_checksums_valid(packet)
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

        # out2in
        pkts = []
        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
             TCP())
        pkts.append(p)
        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
             UDP())
        pkts.append(p)
        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
             ICMPv6EchoReply())
        pkts.append(p)
        p = (Ether(dst=self.pg1.local_mac, src=self.pg1.remote_mac) /
             IPv6(src=self.pg1.remote_ip6, dst=self.nat_addr) /
             GRE() / IP() / TCP())
        pkts.append(p)
        self.pg1.add_stream(pkts)
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg0.get_capture(len(pkts))
        for packet in capture:
            try:
                self.assertEqual(packet[IPv6].src, self.pg1.remote_ip6)
                self.assertEqual(packet[IPv6].dst, self.pg0.remote_ip6)
                self.assert_packet_checksums_valid(packet)
            except:
                self.logger.error(ppp("Unexpected or invalid packet:", packet))
                raise

        sm = self.vapi.nat66_static_mapping_dump()
        self.assertEqual(len(sm), 1)
        self.assertEqual(sm[0].total_pkts, 8)

    def test_check_no_translate(self):
        """ NAT66 translate only when egress interface is outside interface """
        flags = self.config_flags.NAT_IS_INSIDE
        self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
                                          sw_if_index=self.pg0.sw_if_index)
        self.vapi.nat66_add_del_interface(is_add=1, flags=flags,
                                          sw_if_index=self.pg1.sw_if_index)
        self.vapi.nat66_add_del_static_mapping(
            local_ip_address=self.pg0.remote_ip6,
            external_ip_address=self.nat_addr,
            is_add=1)

        # in2out
        p = (Ether(dst=self.pg0.local_mac, src=self.pg0.remote_mac) /
             IPv6(src=self.pg0.remote_ip6, dst=self.pg1.remote_ip6) /
             UDP())
        self.pg0.add_stream([p])
        self.pg_enable_capture(self.pg_interfaces)
        self.pg_start()
        capture = self.pg1.get_capture(1)
        packet = capture[0]
        try:
            self.assertEqual(packet[IPv6].src, self.pg0.remote_ip6)
            self.assertEqual(packet[IPv6].dst, self.pg1.remote_ip6)
        except:
            self.logger.error(ppp("Unexpected or invalid packet:", packet))
            raise


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