resources/libraries/python/SFC/VerifyPacket.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2017 Cisco and/or its affiliates.
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at:
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14
  15 """
  16 This module defines the common functions.
  17 """
  18
  19 import ipaddress
  20
  21 from scapy.layers.inet import IP, UDP
  22 from scapy.all import Raw
  23 from scapy.utils import rdpcap
  24 from resources.libraries.python.constants import Constants as con
  25 from resources.libraries.python.SFC.SFCConstants import SFCConstants as sfccon
  26 from resources.libraries.python.SFC.TunnelProtocol import VxLAN, VxLANGPE, NSH
  27
  28 from robot.api import logger
  29
  30 def valid_ipv4(ipaddr):
  31     """Check if IP address has the correct IPv4 address format.
  32
  33     :param ipaddr: IP address.
  34     :type ipaddr: str
  35     :returns: True in case of correct IPv4 address format,
  36               otherwise return False.
  37     :rtype: bool
  38     """
  39     try:
  40         ipaddress.IPv4Address(unicode(ipaddr))
  41         return True
  42     except (AttributeError, ipaddress.AddressValueError):
  43         return False
  44
  45
  46 def valid_ipv6(ipaddr):
  47     """Check if IP address has the correct IPv6 address format.
  48
  49     :param ipaddr: IP address.
  50     :type ipaddr: str
  51     :returns: True in case of correct IPv6 address format,
  52               otherwise return False.
  53     :rtype: bool
  54     """
  55     try:
  56         ipaddress.IPv6Address(unicode(ipaddr))
  57         return True
  58     except (AttributeError, ipaddress.AddressValueError):
  59         return False
  60
  61
  62 class VerifyPacket(object):
  63     """Define some functions for the test filed verify."""
  64
  65     @staticmethod
  66     def check_vxlan_protocol(payload_data):
  67         """
  68         verify the vxlan protocol in the payload data.
  69
  70         :param payload_data: the payload data in the packet.
  71         :type payload_data: str
  72         :returns: none
  73         :raises RuntimeError: If the vxlan protocol field verify fails.
  74         """
  75         # get the vxlan packet and check it
  76         vxlan_pkt = VxLAN(payload_data[0:8])
  77         if vxlan_pkt.flags != sfccon.VxLAN_FLAGS:
  78             raise RuntimeError("Unexpected Vxlan flags: {0}".
  79                                format(vxlan_pkt.flags))
  80
  81         if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
  82             raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
  83
  84     @staticmethod
  85     def check_vxlangpe_nsh_protocol(payload_data, test_type):
  86         """
  87         verify the vxlangpe and nsh protocol in the payload data.
  88
  89         :param payload_data: the payload data in the packet.
  90         :param test_type: the functional test type.
  91         :type payload_data: str
  92         :type test_type: str
  93         :returns: none
  94         :raises RuntimeError: If the vxlangpe and nsh protocol
  95                               field verify fails.
  96         """
  97         # get the vxlan-gpe packet and check it
  98         vxlangpe_pkt = VxLANGPE(payload_data[0:8])
  99         if vxlangpe_pkt.flags != sfccon.VxLANGPE_FLAGS:
 100             raise RuntimeError("Unexpected Vxlan-GPE flags: {0}".
 101                                format(vxlangpe_pkt.flags))
 102
 103         if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
 104             raise RuntimeError("next protocol not the NSH")
 105
 106         if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
 107             raise RuntimeError("Unexpected VNI flag: {0}".
 108                                format(vxlangpe_pkt.vni))
 109
 110         # get the NSH packet and check it
 111         nsh_pkt = NSH(payload_data[8:32])
 112         if nsh_pkt.flags != sfccon.NSH_FLAGS:
 113             raise RuntimeError("Unexpected NSH flags: {0}".
 114                                format(nsh_pkt.flags))
 115
 116         if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
 117             raise RuntimeError("NSH length {0} incorrect".
 118                                format(nsh_pkt.length))
 119
 120         if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
 121             raise RuntimeError("NSH MD-Type {0} incorrect".
 122                                format(nsh_pkt.MDtype))
 123
 124         if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
 125             raise RuntimeError("NSH next protocol {0} incorrect".
 126                                format(nsh_pkt.nextproto))
 127
 128         if test_type == "Proxy Outbound" or test_type == "SFF":
 129             expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
 130         else:
 131             expect_nsi = sfccon.NSH_DEFAULT_NSI
 132
 133         nsp_nsi = nsh_pkt.nsp_nsi
 134         nsp = nsp_nsi >> 8
 135         nsi = nsp_nsi & 0x000000FF
 136         if nsp != sfccon.NSH_DEFAULT_NSP:
 137             raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
 138
 139         if nsi != expect_nsi:
 140             raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
 141
 142         nsh_c1 = nsh_pkt.c1
 143         if nsh_c1 != sfccon.NSH_DEFAULT_C1:
 144             raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
 145
 146         nsh_c2 = nsh_pkt.c2
 147         if nsh_c2 != sfccon.NSH_DEFAULT_C2:
 148             raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
 149
 150         nsh_c3 = nsh_pkt.c3
 151         if nsh_c3 != sfccon.NSH_DEFAULT_C3:
 152             raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
 153
 154         nsh_c4 = nsh_pkt.c4
 155         if nsh_c4 != sfccon.NSH_DEFAULT_C4:
 156             raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
 157
 158
 159     @staticmethod
 160     def check_the_nsh_sfc_packet(frame_size, test_type):
 161         """
 162         verify the NSH SFC functional test loopback packet field
 163         is correct.
 164
 165         :param frame_size: the origin frame size.
 166         :param test_type: the test type.
 167                          (Classifier, Proxy Inbound, Proxy Outbound, SFF).
 168         :type frame_size: Integer
 169         :type test_type: str
 170         :returns: none
 171         :raises RuntimeError: If the packet field verify fails.
 172         """
 173
 174         rx_pcapfile = '{0}/nsh_sfc_tests/sfc_scripts/temp_packet.pcap' \
 175                       .format(con.REMOTE_FW_DIR)
 176
 177         logger.trace('read pcap file:{0}'.format(rx_pcapfile))
 178
 179         packets = rdpcap(rx_pcapfile)
 180         if len(packets) < 1:
 181             raise RuntimeError("No packet is received!")
 182
 183         ether = packets[0]
 184
 185         origin_size = int(frame_size)
 186         if test_type == "Classifier":
 187             expect_pkt_len = origin_size + 74 - 4
 188         elif test_type == "Proxy Inbound":
 189             expect_pkt_len = origin_size - 24 - 4
 190         elif test_type == "Proxy Outbound":
 191             expect_pkt_len = origin_size + 24 - 4
 192         else:
 193             expect_pkt_len = origin_size - 4
 194
 195         recv_pkt_len = len(ether)
 196         if recv_pkt_len != expect_pkt_len:
 197             raise RuntimeError("Received packet size {0} not " \
 198                                "the expect size {1}".format(recv_pkt_len, \
 199                                expect_pkt_len))
 200
 201         if not ether.haslayer(IP):
 202             raise RuntimeError("Not a IPv4 packet")
 203
 204         pkt_proto = ether[IP].proto
 205         if pkt_proto != sfccon.UDP_PROTOCOL:
 206             raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
 207
 208         if test_type == "Proxy Inbound":
 209             expect_udp_port = sfccon.VxLAN_UDP_PORT
 210         else:
 211             expect_udp_port = sfccon.VxLANGPE_UDP_PORT
 212
 213         dst_port = ether[UDP].dport
 214         if dst_port != expect_udp_port:
 215             raise RuntimeError("UDP dest port must be {0}, {1}".
 216                                format(expect_udp_port, dst_port))
 217
 218         payload_data = ether[Raw].load
 219
 220         if test_type == "Proxy Inbound":
 221             VerifyPacket.check_vxlan_protocol(payload_data)
 222         else:
 223             VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)