path: root/scripts/automation/regression/functional_tests/scapy_pkt_builder_test.py

#!/router/bin/python

import pkt_bld_general_test

#HACK FIX ME START
import sys
import os

CURRENT_PATH        = os.path.dirname(os.path.realpath(__file__))
sys.path.append(os.path.join(CURRENT_PATH, '../../../trex_control_plane/stl/'))
#HACK FIX ME END
from trex_stl_lib.trex_stl_packet_builder_scapy import *

from scapy.all import *
from nose.tools import assert_equal
from nose.tools import assert_not_equal
from nose.tools import assert_raises
from nose.tools import raises
import os
import random
import pprint

class CTRexPktBuilderSanitySCapy_Test(pkt_bld_general_test.CGeneralPktBld_Test):

    def setUp(self):
        pass

    def test_simple_vm1(self):
        raw1 = STLScVmRaw( [ STLVmFlowVar(name="a",min_value="16.0.0.1",max_value="16.0.0.10",init_value="16.0.0.1",size=4,op="inc"),
                             STLVmWrFlowVar(fv_name="a",pkt_offset= "IP.src"),
                             STLVmFixIpv4(offset = "IP")]
                          );

        pkt_builder = STLPktBuilder();

        py='5'*128
        pkt=Ether()/ \
                 IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)/IP()/py

        # set packet 
        pkt_builder.set_packet(pkt);
        pkt_builder.add_command ( raw1 )
        pkt_builder.compile();

        pkt_builder.dump_scripts ()

        print(pkt_builder.get_vm_data())

        assert_equal( pkt_builder.get_vm_data(), {'split_by_var': '', 'instructions': [{'name': 'a', 'max_value': 268435466, 'min_value': 268435457, 'init_value': 268435457, 'size': 4, 'type': 'flow_var', 'step':1,'op': 'inc'}, {'is_big_endian': True, 'pkt_offset': 26, 'type': 'write_flow_var',  'name': 'a', 'add_value': 0}, {'pkt_offset': 14, 'type': 'fix_checksum_ipv4'}]} )
                                                 


    def test_simple_no_vm1(self):

        pkt_builder = STLPktBuilder();

        py='5'*128
        pkt=Ether()/ \
                 IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)/IP()/py

        # set packet 
        pkt_builder.set_packet(pkt);

        pkt_builder.compile();

        pkt_builder.dump_scripts ()

        assert_equal( pkt_builder.get_vm_data(),
                {   'instructions': [ ],
                    'split_by_var': ''}
        )


    def test_simple_mac_default(self):

        pkt =  Ether()/IP()/UDP()


        pkt_builder = STLPktBuilder(pkt = pkt);

        assert_equal( pkt_builder.is_default_src_mac () ,True)
        assert_equal( pkt_builder.is_default_dst_mac () ,True)

        pkt =  Ether(src="00:00:00:00:00:01")/IP()/UDP()

        pkt_builder = STLPktBuilder(pkt = pkt);

        assert_equal( pkt_builder.is_default_src_mac (), False)
        assert_equal( pkt_builder.is_default_dst_mac (), True)

        pkt =  Ether(dst="00:00:00:00:00:01")/IP()/UDP()

        pkt_builder = STLPktBuilder(pkt = pkt);

        assert_equal( pkt_builder.is_default_src_mac (),True)
        assert_equal(  pkt_builder.is_default_dst_mac (),False)




    def test_simple_teredo(self):

        pkt =  Ether()/IP(src="16.0.0.1",dst="48.0.0.1")/UDP(dport=3797,sport=3544)/IPv6(src="2001:0:4137:9350:8000:f12a:b9c8:2815",dst="2001:4860:0:2001::68")/UDP(dport=12,sport=1025)/ICMPv6Unknown()

        pkt.build();
        p_utl=CTRexScapyPktUtl(pkt);

        assert_equal( p_utl.get_field_offet_by_str("IPv6.src"), (50,16) )
        assert_equal( p_utl.get_field_offet_by_str("IPv6.dst"), (66,16) )




    def test_simple_scapy_vlan(self):

        py='5'*(9)
        p1=Ether(src="00:00:00:01:00:00",dst="00:00:00:01:00:00")/ \
                 Dot1Q(vlan=12)/ \
                 Dot1Q(vlan=17)/ \
                 IP(src="10.0.0.10",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)/py

        p1.build();
        p1.dump_layers_offset()
        p1.show2();
        hexdump(p1);
        #wrpcap("ipv4_udp_9k.pcap", p1);

        p_utl=CTRexScapyPktUtl(p1);

        assert_equal(p_utl.get_pkt_layers(),"Ethernet:802.1Q:802.1Q:IP:UDP:Raw")
        assert_equal(p_utl.layer_offset("802.1Q",0),14);
        assert_equal(p_utl.layer_offset("802.1Q",1),18);
        assert_equal(p_utl.get_field_offet_by_str("802|1Q.vlan"),(14,0));
        assert_equal(p_utl.get_field_offet_by_str("802|1Q:1.vlan"),(18,0));
        assert_equal(p_utl.get_field_offet_by_str("IP.src"),(34,4));

    def test_simple_scapy_128_udp(self):
        """
        build 128 byte packet with 0x35 as pyld
        """


        pkt_size =128 
        p1=Ether(src="00:00:00:01:00:00",dst="00:00:00:01:00:00")/ \
                 IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)
        pyld_size=pkt_size-len(p1);

        pkt=p1/('5'*(pyld_size))

        pkt.show2();
        hexdump(pkt);
        assert_equal(len(pkt),128)

    def test_simple_scapy_9k_ip_len(self):
        """
        build 9k ipv4 len packet
        """


        ip_pkt_size =9*1024
        p_l2=Ether(src="00:00:00:01:00:00",dst="00:00:00:01:00:00");
        p_l3=    IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)
        pyld_size = ip_pkt_size-len(p_l3);

        pkt=p_l2/p_l3/('\x55'*(pyld_size))

        #pkt.show2();
        #hexdump(pkt);
        assert_equal(len(pkt),9*1024+14)

    def test_simple_scapy_ipv6_1(self):
        """
        build ipv6 packet 
        """

        print("start ")
        py='\x55'*(64)

        p=Ether()/IPv6()/UDP(dport=12,sport=1025)/py
        #p.build();
        #p.dump_layers_offset()
        hexdump(p);
        p.show2();

        p_utl=CTRexScapyPktUtl(p);

        assert_equal(p_utl.get_field_offet_by_str("IPv6.src"),(22,16));


    def test_simple_vm2(self):
        raw1 = STLScVmRaw( [ STLVmFlowVar(name="my_valn",min_value=0,max_value=10,init_value=2,size=1,op="inc"),
                             STLVmWrFlowVar (fv_name="my_valn",pkt_offset= "802|1Q.vlan" ,offset_fixup=3) # fix the offset as valn is bitfield and not supported right now 
                              ]
                          );

        pkt_builder = STLPktBuilder();

        py='5'*128
        pkt=Ether()/ \
        Dot1Q(vlan=12)/ \
                 IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)/IP()/py

        # set packet 
        pkt_builder.set_packet(pkt);
        pkt_builder.add_command ( raw1 )
        pkt_builder.compile();


        d= pkt_builder.get_vm_data()
        assert_equal(d['instructions'][1]['pkt_offset'],17)

    def test_simple_vm3(self):
        try:
            raw1 = STLScVmRaw( [ STLVmFlowVar(name="my_valn",min_value=0,max_value=10,init_value=2,size=1,op="inc"),
                                 STLVmWrFlowVar(fv_name="my_valn_err",pkt_offset= "802|1Q.vlan" ,offset_fixup=3) # fix the offset as valn is bitfield and not supported right now 
                                  ]
                              );
    
            pkt_builder = STLPktBuilder();
    
            py='5'*128
            pkt=Ether()/ \
            Dot1Q(vlan=12)/ \
                     IP(src="16.0.0.1",dst="48.0.0.1")/ \
                     UDP(dport=12,sport=1025)/IP()/py
    
            # set packet 
            pkt_builder.set_packet(pkt);
            pkt_builder.add_command ( raw1 )
            pkt_builder.compile();
    
    
            d= pkt_builder.get_vm_data()
        except  CTRexPacketBuildException as e:
            error=str(e) 
            assert_equal(error.find("[errcode:-11]"),0);

    def test_simple_tuple_gen(self):
        vm = STLScVmRaw( [ STLVmTupleGen(name="tuple"), # define tuple gen 
                           STLVmWrFlowVar(fv_name="tuple.ip", pkt_offset= "IP.src" ), # write ip to packet IP.src
                           STLVmFixIpv4(offset = "IP"),                                # fix checksum
                           STLVmWrFlowVar (fv_name="tuple.port", pkt_offset= "UDP.sport" )  #write udp.port
                          ]
                        );
        pkt_builder = STLPktBuilder();

        py='5'*128
        pkt=Ether()/ \
        Dot1Q(vlan=12)/ \
                 IP(src="16.0.0.1",dst="48.0.0.1")/ \
                 UDP(dport=12,sport=1025)/IP()/py

        # set packet 
        pkt_builder.set_packet(pkt);
        pkt_builder.add_command ( vm )
        pkt_builder.compile();
        d= pkt_builder.get_vm_data()
        pkt_builder.dump_vm_data_as_yaml()

        assert_equal(d['instructions'][1]['pkt_offset'],30)
        assert_equal(d['instructions'][3]['pkt_offset'],38)

    def test_simple_random_pkt_size(self):

        ip_pkt_size = 9*1024
        p_l2 = Ether();
        p_l3 = IP(src="16.0.0.1",dst="48.0.0.1")
        p_l4 = UDP(dport=12,sport=1025)
        pyld_size = ip_pkt_size-len(p_l3/p_l4);

        pkt =p_l2/p_l3/p_l4/('\x55'*(pyld_size))

        l3_len_fix =-(len(p_l2));
        l4_len_fix =-(len(p_l2/p_l3));

        vm = STLScVmRaw( [ STLVmFlowVar(name="fv_rand", min_value=64, max_value=len(pkt), size=2, op="random"),
                           STLVmTrimPktSize("fv_rand"), # total packet size
                           STLVmWrFlowVar(fv_name="fv_rand", pkt_offset= "IP.len", add_val=l3_len_fix), 
                           STLVmFixIpv4(offset = "IP"),                                # fix checksum
                           STLVmWrFlowVar(fv_name="fv_rand", pkt_offset= "UDP.len", add_val=l4_len_fix)  
                          ]
                       )
        pkt_builder = STLPktBuilder();

        # set packet 
        pkt_builder.set_packet(pkt);
        pkt_builder.add_command ( vm )
        pkt_builder.compile();
        d= pkt_builder.get_vm_data()
        pkt_builder.dump_vm_data_as_yaml()

        assert_equal(d['instructions'][0]['max_value'],9230)
        assert_equal(d['instructions'][2]['pkt_offset'],16)
        assert_equal(d['instructions'][4]['pkt_offset'],38)

    def test_simple_pkt_loader(self):
        p=RawPcapReader("functional_tests/golden/basic_imix_golden.cap")
        print("")
        for pkt in p:
            print(pkt[1])
            print(hexdump(str(pkt[0])))
            break;

    def test_simple_pkt_loader1(self):

        pkt_builder = STLPktBuilder(pkt = "functional_tests/golden/udp_590.cap", build_raw = False);
        print("")
        pkt_builder.dump_as_hex()
        r = pkt_builder.pkt_raw
        assert_equal(safe_ord(r[1]),0x50)
        assert_equal(safe_ord(r[0]),0x00)
        assert_equal(safe_ord(r[0x240]),0x16)
        assert_equal(safe_ord(r[0x24d]),0x79)
        assert_equal(len(r),590)

        print(len(r))

    def test_simple_pkt_loader2(self):

        pkt_builder = STLPktBuilder(pkt = "functional_tests/golden/basic_imix_golden.cap");
        assert_equal(pkt_builder.pkt_layers_desc (), "Ethernet:IP:UDP:Raw");

    def test_simple_pkt_loader3(self):

        #pkt_builder = STLPktBuilder(pkt = "stl/golden/basic_imix_golden.cap");
        #r = pkt_builder.pkt_raw
        #print ""
        #hexdump(str(r))


        #print pkt_builder.pkt_layers_desc ()


        #pkt_builder.set_packet(pkt);

        py='\x55'*(64)

        p=Ether()/IP()/UDP(dport=12,sport=1025)/py
        pkt_str = bytes(p);
        print("")
        hexdump(pkt_str);
        scapy_pkt = Ether(pkt_str);
        scapy_pkt.show2();

    def tearDown(self):
        pass


class CTRexPktBuilderScapy_Test(pkt_bld_general_test.CGeneralPktBld_Test):

    def setUp(self):
        pass;
        #self.pkt_bld = CTRexPktBuilder()
        #self.pkt_bld.add_pkt_layer("l2", dpkt.ethernet.Ethernet())
        #self.pp = pprint.PrettyPrinter(indent=4)

    def tearDown(self):
        pass


if __name__ == "__main__":
    pass