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 resources.libraries.python.SFC.SFCConstants import SFCConstants as sfccon
  24 from resources.libraries.python.SFC.TunnelProtocol import VxLAN, VxLANGPE, NSH
  25
  26
  27 def valid_ipv4(ipaddr):
  28     """Check if IP address has the correct IPv4 address format.
  29
  30     :param ipaddr: IP address.
  31     :type ipaddr: str
  32     :returns: True in case of correct IPv4 address format,
  33               otherwise return False.
  34     :rtype: bool
  35     """
  36     try:
  37         ipaddress.IPv4Address(unicode(ipaddr))
  38         return True
  39     except (AttributeError, ipaddress.AddressValueError):
  40         return False
  41
  42
  43 def valid_ipv6(ipaddr):
  44     """Check if IP address has the correct IPv6 address format.
  45
  46     :param ipaddr: IP address.
  47     :type ipaddr: str
  48     :returns: True in case of correct IPv6 address format,
  49               otherwise return False.
  50     :rtype: bool
  51     """
  52     try:
  53         ipaddress.IPv6Address(unicode(ipaddr))
  54         return True
  55     except (AttributeError, ipaddress.AddressValueError):
  56         return False
  57
  58
  59 class VerifyPacket(object):
  60     """Define some functions for the test filed verify."""
  61
  62     @staticmethod
  63     def check_vxlan_protocol(payload_data):
  64         """
  65         verify the vxlan protocol in the payload data.
  66
  67         :param payload_data: the payload data in the packet.
  68         :type payload_data: str
  69         :raises RuntimeError: If the vxlan protocol field verify fails.
  70         """
  71         # get the vxlan packet and check it
  72         vxlan_pkt = VxLAN(payload_data[0:8])
  73         if vxlan_pkt.flags != sfccon.VxLAN_FLAGS:
  74             raise RuntimeError("Unexpected Vxlan flags: {0}".
  75                                format(vxlan_pkt.flags))
  76
  77         if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
  78             raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
  79
  80     @staticmethod
  81     def check_vxlangpe_nsh_protocol(payload_data, test_type):
  82         """
  83         verify the vxlangpe and nsh protocol in the payload data.
  84
  85         :param payload_data: the payload data in the packet.
  86         :param test_type: the functional test type.
  87         :type payload_data: str
  88         :type test_type: str
  89         :raises RuntimeError: If the vxlangpe and nsh protocol
  90                               field verify fails.
  91         """
  92         # get the vxlan-gpe packet and check it
  93         vxlangpe_pkt = VxLANGPE(payload_data[0:8])
  94         if vxlangpe_pkt.flags != sfccon.VxLANGPE_FLAGS:
  95             raise RuntimeError("Unexpected Vxlan-GPE flags: {0}".
  96                                format(vxlangpe_pkt.flags))
  97
  98         if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
  99             raise RuntimeError("next protocol not the NSH")
 100
 101         if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
 102             raise RuntimeError("Unexpected VNI flag: {0}".
 103                                format(vxlangpe_pkt.vni))
 104
 105         # get the NSH packet and check it
 106         nsh_pkt = NSH(payload_data[8:32])
 107         if nsh_pkt.Version != 0:
 108             raise RuntimeError("Unexpected NSH version: {0}".
 109                                format(nsh_pkt.Version))
 110
 111         if nsh_pkt.OAM != 0 and nsh_pkt.OAM != 1:
 112             raise RuntimeError("Unexpected NSH OAM: {0}".
 113                                format(nsh_pkt.OAM))
 114
 115         if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
 116             raise RuntimeError("NSH length {0} incorrect".
 117                                format(nsh_pkt.length))
 118
 119         if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
 120             raise RuntimeError("NSH MD-Type {0} incorrect".
 121                                format(nsh_pkt.MDtype))
 122
 123         if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
 124             raise RuntimeError("NSH next protocol {0} incorrect".
 125                                format(nsh_pkt.nextproto))
 126
 127         if test_type == "Proxy Outbound" or test_type == "SFF":
 128             expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
 129         else:
 130             expect_nsi = sfccon.NSH_DEFAULT_NSI
 131
 132         nsp_nsi = nsh_pkt.nsp_nsi
 133         nsp = nsp_nsi >> 8
 134         nsi = nsp_nsi & 0x000000FF
 135         if nsp != sfccon.NSH_DEFAULT_NSP:
 136             raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
 137
 138         if nsi != expect_nsi:
 139             raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
 140
 141         nsh_c1 = nsh_pkt.c1
 142         if nsh_c1 != sfccon.NSH_DEFAULT_C1:
 143             raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
 144
 145         nsh_c2 = nsh_pkt.c2
 146         if nsh_c2 != sfccon.NSH_DEFAULT_C2:
 147             raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
 148
 149         nsh_c3 = nsh_pkt.c3
 150         if nsh_c3 != sfccon.NSH_DEFAULT_C3:
 151             raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
 152
 153         nsh_c4 = nsh_pkt.c4
 154         if nsh_c4 != sfccon.NSH_DEFAULT_C4:
 155             raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
 156
 157
 158     @staticmethod
 159     def check_the_nsh_sfc_packet(ether, frame_size, test_type):
 160         """
 161         verify the NSH SFC functional test loopback packet field
 162         is correct.
 163
 164         :param ether: The Ethernet packet data.
 165         :param frame_size: The origin frame size.
 166         :param test_type: The test type.
 167                          (Classifier, Proxy Inbound, Proxy Outbound, SFF).
 168
 169         :type ether: scapy.Ether
 170         :type frame_size: Integer
 171         :type test_type: str
 172         :raises RuntimeError: If the packet field verify fails.
 173         """
 174
 175         origin_size = int(frame_size)
 176         if test_type == "Classifier":
 177             expect_pkt_len = origin_size + 74 - 4
 178         elif test_type == "Proxy Inbound":
 179             expect_pkt_len = origin_size - 24 - 4
 180         elif test_type == "Proxy Outbound":
 181             expect_pkt_len = origin_size + 24 - 4
 182         else:
 183             expect_pkt_len = origin_size - 4
 184
 185         recv_pkt_len = len(ether)
 186         if recv_pkt_len != expect_pkt_len:
 187             raise RuntimeError("Received packet size {0} not "
 188                                "the expect size {1}".format(recv_pkt_len,
 189                                                             expect_pkt_len))
 190
 191         if not ether.haslayer(IP):
 192             raise RuntimeError("Not a IPv4 packet")
 193
 194         pkt_proto = ether[IP].proto
 195         if pkt_proto != sfccon.UDP_PROTOCOL:
 196             raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
 197
 198         if test_type == "Proxy Inbound":
 199             expect_udp_port = sfccon.VxLAN_UDP_PORT
 200         else:
 201             expect_udp_port = sfccon.VxLANGPE_UDP_PORT
 202
 203         dst_port = ether[UDP].dport
 204         if dst_port != expect_udp_port:
 205             raise RuntimeError("UDP dest port must be {0}, {1}".
 206                                format(expect_udp_port, dst_port))
 207
 208         payload_data = ether[Raw].load
 209
 210         if test_type == "Proxy Inbound":
 211             VerifyPacket.check_vxlan_protocol(payload_data)
 212         else:
 213             VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)