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         :returns: none
  70         :raises RuntimeError: If the vxlan protocol field verify fails.
  71         """
  72         # get the vxlan packet and check it
  73         vxlan_pkt = VxLAN(payload_data[0:8])
  74         if vxlan_pkt.flags != sfccon.VxLAN_FLAGS:
  75             raise RuntimeError("Unexpected Vxlan flags: {0}".
  76                                format(vxlan_pkt.flags))
  77
  78         if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
  79             raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
  80
  81     @staticmethod
  82     def check_vxlangpe_nsh_protocol(payload_data, test_type):
  83         """
  84         verify the vxlangpe and nsh protocol in the payload data.
  85
  86         :param payload_data: the payload data in the packet.
  87         :param test_type: the functional test type.
  88         :type payload_data: str
  89         :type test_type: str
  90         :returns: none
  91         :raises RuntimeError: If the vxlangpe and nsh protocol
  92                               field verify fails.
  93         """
  94         # get the vxlan-gpe packet and check it
  95         vxlangpe_pkt = VxLANGPE(payload_data[0:8])
  96         if vxlangpe_pkt.flags != sfccon.VxLANGPE_FLAGS:
  97             raise RuntimeError("Unexpected Vxlan-GPE flags: {0}".
  98                                format(vxlangpe_pkt.flags))
  99
 100         if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
 101             raise RuntimeError("next protocol not the NSH")
 102
 103         if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
 104             raise RuntimeError("Unexpected VNI flag: {0}".
 105                                format(vxlangpe_pkt.vni))
 106
 107         # get the NSH packet and check it
 108         nsh_pkt = NSH(payload_data[8:32])
 109         if nsh_pkt.flags != sfccon.NSH_FLAGS:
 110             raise RuntimeError("Unexpected NSH flags: {0}".
 111                                format(nsh_pkt.flags))
 112
 113         if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
 114             raise RuntimeError("NSH length {0} incorrect".
 115                                format(nsh_pkt.length))
 116
 117         if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
 118             raise RuntimeError("NSH MD-Type {0} incorrect".
 119                                format(nsh_pkt.MDtype))
 120
 121         if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
 122             raise RuntimeError("NSH next protocol {0} incorrect".
 123                                format(nsh_pkt.nextproto))
 124
 125         if test_type == "Proxy Outbound" or test_type == "SFF":
 126             expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
 127         else:
 128             expect_nsi = sfccon.NSH_DEFAULT_NSI
 129
 130         nsp_nsi = nsh_pkt.nsp_nsi
 131         nsp = nsp_nsi >> 8
 132         nsi = nsp_nsi & 0x000000FF
 133         if nsp != sfccon.NSH_DEFAULT_NSP:
 134             raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
 135
 136         if nsi != expect_nsi:
 137             raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
 138
 139         nsh_c1 = nsh_pkt.c1
 140         if nsh_c1 != sfccon.NSH_DEFAULT_C1:
 141             raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
 142
 143         nsh_c2 = nsh_pkt.c2
 144         if nsh_c2 != sfccon.NSH_DEFAULT_C2:
 145             raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
 146
 147         nsh_c3 = nsh_pkt.c3
 148         if nsh_c3 != sfccon.NSH_DEFAULT_C3:
 149             raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
 150
 151         nsh_c4 = nsh_pkt.c4
 152         if nsh_c4 != sfccon.NSH_DEFAULT_C4:
 153             raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
 154
 155
 156     @staticmethod
 157     def check_the_nsh_sfc_packet(ether, frame_size, test_type):
 158         """
 159         verify the NSH SFC functional test loopback packet field
 160         is correct.
 161
 162         :param ether: The Ethernet packet data.
 163         :param frame_size: The origin frame size.
 164         :param test_type: The test type.
 165                          (Classifier, Proxy Inbound, Proxy Outbound, SFF).
 166
 167         :type ether: scapy.Ether
 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         origin_size = int(frame_size)
 175         if test_type == "Classifier":
 176             expect_pkt_len = origin_size + 74 - 4
 177         elif test_type == "Proxy Inbound":
 178             expect_pkt_len = origin_size - 24 - 4
 179         elif test_type == "Proxy Outbound":
 180             expect_pkt_len = origin_size + 24 - 4
 181         else:
 182             expect_pkt_len = origin_size - 4
 183
 184         recv_pkt_len = len(ether)
 185         if recv_pkt_len != expect_pkt_len:
 186             raise RuntimeError("Received packet size {0} not "
 187                                "the expect size {1}".format(recv_pkt_len,
 188                                                             expect_pkt_len))
 189
 190         if not ether.haslayer(IP):
 191             raise RuntimeError("Not a IPv4 packet")
 192
 193         pkt_proto = ether[IP].proto
 194         if pkt_proto != sfccon.UDP_PROTOCOL:
 195             raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
 196
 197         if test_type == "Proxy Inbound":
 198             expect_udp_port = sfccon.VxLAN_UDP_PORT
 199         else:
 200             expect_udp_port = sfccon.VxLANGPE_UDP_PORT
 201
 202         dst_port = ether[UDP].dport
 203         if dst_port != expect_udp_port:
 204             raise RuntimeError("UDP dest port must be {0}, {1}".
 205                                format(expect_udp_port, dst_port))
 206
 207         payload_data = ether[Raw].load
 208
 209         if test_type == "Proxy Inbound":
 210             VerifyPacket.check_vxlan_protocol(payload_data)
 211         else:
 212             VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)