#!/usr/bin/env python import socket import unittest import struct from framework import VppTestCase, VppTestRunner, running_extended_tests from scapy.layers.inet import IP, TCP, UDP, ICMP from scapy.layers.inet import IPerror, TCPerror, UDPerror, ICMPerror from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply from scapy.layers.inet6 import ICMPv6DestUnreach, IPerror6 from scapy.layers.l2 import Ether, ARP, GRE from scapy.data import IP_PROTOS from scapy.packet import bind_layers from util import ppp from ipfix import IPFIX, Set, Template, Data, IPFIXDecoder from time import sleep from util import ip4_range from util import mactobinary class MethodHolder(VppTestCase): """ NAT create capture and verify method holder """ @classmethod def setUpClass(cls): super(MethodHolder, cls).setUpClass() def tearDown(self): super(MethodHolder, self).tearDown() def check_ip_checksum(self, pkt): """ Check IP checksum of the packet :param pkt: Packet to check IP checksum """ new = pkt.__class__(str(pkt)) del new['IP'].chksum new = new.__class__(str(new)) self.assertEqual(new['IP'].chksum, pkt['IP'].chksum) def check_tcp_checksum(self, pkt): """ Check TCP checksum in IP packet :param pkt: Packet to check TCP checksum """ new = pkt.__class__(str(pkt)) del new['TCP'].chksum new = new.__class__(str(new)) self.assertEqual(new['TCP'].chksum, pkt['TCP'].chksum) def check_udp_checksum(self, pkt): """ Check UDP checksum in IP packet :param pkt: Packet to check UDP checksum """ new = pkt.__class__(str(pkt)) del new['UDP'].chksum new = new.__class__(str(new)) self.assertEqual(new['UDP'].chksum, pkt['UDP'].chksum) def check_icmp_errror_embedded(self, pkt): """ Check ICMP error embeded packet checksum :param pkt: Packet to check ICMP error embeded packet checksum """ if pkt.haslayer(IPerror): new = pkt.__class__(str(pkt)) del new['IPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['IPerror'].chksum, pkt['IPerror'].chksum) if pkt.haslayer(TCPerror): new = pkt.__class__(str(pkt)) del new['TCPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['TCPerror'].chksum, pkt['TCPerror'].chksum) if pkt.haslayer(UDPerror): if pkt['UDPerror'].chksum != 0: new = pkt.__class__(str(pkt)) del new['UDPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['UDPerror'].chksum, pkt['UDPerror'].chksum) if pkt.haslayer(ICMPerror): del new['ICMPerror'].chksum new = new.__class__(str(new)) self.assertEqual(new['ICMPerror'].chksum, pkt['ICMPerror'].chksum) def check_icmp_checksum(self, pkt): """ Check ICMP checksum in IPv4 packet :param pkt: Packet to check ICMP checksum """ new = pkt.__class__(str(pkt)) del new['ICMP'].chksum new = new.__class__(str(new)) self.assertEqual(new['ICMP'].chksum, pkt['ICMP'].chksum) if pkt.haslayer(IPerror): self.check_icmp_errror_embedded(pkt) def check_icmpv6_checksum(self, pkt): """ Check ICMPv6 checksum in IPv4 packet :param pkt: Packet to check ICMPv6 checksum """ new = pkt.__class__(str(pkt)) if pkt.haslayer(ICMPv6DestUnreach): del new['ICMPv6DestUnreach'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6DestUnreach'].cksum, pkt['ICMPv6DestUnreach'].cksum) self.check_icmp_errror_embedded(pkt) if pkt.haslayer(ICMPv6EchoRequest): del new['ICMPv6EchoRequest'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6EchoRequest'].cksum, pkt['ICMPv6EchoRequest'].cksum) if pkt.haslayer(ICMPv6EchoReply): del new['ICMPv6EchoReply'].cksum new = new.__class__(str(new)) self.assertEqual(new['ICMPv6EchoReply'].cksum, pkt['ICMPv6EchoReply'].cksum) def create_stream_in(self, in_if, out_if, ttl=64): """ Create packet stream for inside network :param in_if: Inside interface :param out_if: Outside interface :param ttl: TTL of generated packets """ pkts = [] # TCP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / TCP(sport=self.tcp_port_in, dport=20)) pkts.append(p) # UDP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / UDP(sport=self.udp_port_in, dport=20)) pkts.append(p) # ICMP p = (Ether(dst=in_if.local_mac, src=in_if.remote_mac) / IP(src=in_if.remote_ip4, dst=out_if.remote_ip4, ttl=ttl) / ICMP(id=self.icmp_id_in, type='echo-request')) pkts.append(p) return pkts def compose_ip6(self, ip4, pref, plen): """ Compose IPv4-embedded IPv6 addresses :param ip4: IPv4 address :param pref: IPv6 prefix :param plen: IPv6 prefix length :returns: IPv4-embedded IPv6 addresses """ pre
set int ip address GigabitEthernet2/1/0 1.2.3.4/24
set int state GigabitEthernet2/1/0 up
set int state GigabitEthernet2/2/0 up
cre sub GigabitEthernet2/2/0 1 dot1q 9
set int state GigabitEthernet2/2/0.1 up
set int state tuntap-0 down
set int acl input GigabitEthernet2/1/0
set int acl output GigabitEthernet2/1/0
set int acl input GigabitEthernet2/2/0.1
set int acl output GigabitEthernet2/2/0.1
set int l2 xc GigabitEthernet2/1/0 GigabitEthernet2/2/0.1
set int l2 xc GigabitEthernet2/2/0.1 GigabitEthernet2/1/0
clear error
clear run
clear int
packet-generator new {
name xc
limit 11
no-recycle
node ethernet-input
interface GigabitEthernet2/1/0
size 98-98
data { hex 0x00010203040500020406080a080045006402b46b96000100096978676265000500bf436973636f20494f5320536f6674776172652c2040160011000000010101cc000400000000001a00100000000100000000ffffffff }
}