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:
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
16 This module defines the common functions.
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
27 def valid_ipv4(ipaddr):
28 """Check if IP address has the correct IPv4 address format.
30 :param ipaddr: IP address.
32 :returns: True in case of correct IPv4 address format,
33 otherwise return False.
37 ipaddress.IPv4Address(unicode(ipaddr))
39 except (AttributeError, ipaddress.AddressValueError):
43 def valid_ipv6(ipaddr):
44 """Check if IP address has the correct IPv6 address format.
46 :param ipaddr: IP address.
48 :returns: True in case of correct IPv6 address format,
49 otherwise return False.
53 ipaddress.IPv6Address(unicode(ipaddr))
55 except (AttributeError, ipaddress.AddressValueError):
59 class VerifyPacket(object):
60 """Define some functions for the test filed verify."""
63 def check_vxlan_protocol(payload_data):
65 verify the vxlan protocol in the payload data.
67 :param payload_data: the payload data in the packet.
68 :type payload_data: str
70 :raises RuntimeError: If the vxlan protocol field verify fails.
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))
78 if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
79 raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
82 def check_vxlangpe_nsh_protocol(payload_data, test_type):
84 verify the vxlangpe and nsh protocol in the payload data.
86 :param payload_data: the payload data in the packet.
87 :param test_type: the functional test type.
88 :type payload_data: str
91 :raises RuntimeError: If the vxlangpe and nsh protocol
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))
100 if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
101 raise RuntimeError("next protocol not the NSH")
103 if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
104 raise RuntimeError("Unexpected VNI flag: {0}".
105 format(vxlangpe_pkt.vni))
107 # get the NSH packet and check it
108 nsh_pkt = NSH(payload_data[8:32])
109 if nsh_pkt.Version != 0:
110 raise RuntimeError("Unexpected NSH version: {0}".
111 format(nsh_pkt.Version))
113 print type(nsh_pkt.OAM)
114 if nsh_pkt.OAM != 0 or nsh_pkt.OAM != 1:
115 raise RuntimeError("Unexpected NSH OAM: {0}".
118 if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
119 raise RuntimeError("NSH length {0} incorrect".
120 format(nsh_pkt.length))
122 if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
123 raise RuntimeError("NSH MD-Type {0} incorrect".
124 format(nsh_pkt.MDtype))
126 if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
127 raise RuntimeError("NSH next protocol {0} incorrect".
128 format(nsh_pkt.nextproto))
130 if test_type == "Proxy Outbound" or test_type == "SFF":
131 expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
133 expect_nsi = sfccon.NSH_DEFAULT_NSI
135 nsp_nsi = nsh_pkt.nsp_nsi
137 nsi = nsp_nsi & 0x000000FF
138 if nsp != sfccon.NSH_DEFAULT_NSP:
139 raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
141 if nsi != expect_nsi:
142 raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
145 if nsh_c1 != sfccon.NSH_DEFAULT_C1:
146 raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
149 if nsh_c2 != sfccon.NSH_DEFAULT_C2:
150 raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
153 if nsh_c3 != sfccon.NSH_DEFAULT_C3:
154 raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
157 if nsh_c4 != sfccon.NSH_DEFAULT_C4:
158 raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
162 def check_the_nsh_sfc_packet(ether, frame_size, test_type):
164 verify the NSH SFC functional test loopback packet field
167 :param ether: The Ethernet packet data.
168 :param frame_size: The origin frame size.
169 :param test_type: The test type.
170 (Classifier, Proxy Inbound, Proxy Outbound, SFF).
172 :type ether: scapy.Ether
173 :type frame_size: Integer
176 :raises RuntimeError: If the packet field verify fails.
179 origin_size = int(frame_size)
180 if test_type == "Classifier":
181 expect_pkt_len = origin_size + 74 - 4
182 elif test_type == "Proxy Inbound":
183 expect_pkt_len = origin_size - 24 - 4
184 elif test_type == "Proxy Outbound":
185 expect_pkt_len = origin_size + 24 - 4
187 expect_pkt_len = origin_size - 4
189 recv_pkt_len = len(ether)
190 if recv_pkt_len != expect_pkt_len:
191 raise RuntimeError("Received packet size {0} not "
192 "the expect size {1}".format(recv_pkt_len,
195 if not ether.haslayer(IP):
196 raise RuntimeError("Not a IPv4 packet")
198 pkt_proto = ether[IP].proto
199 if pkt_proto != sfccon.UDP_PROTOCOL:
200 raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
202 if test_type == "Proxy Inbound":
203 expect_udp_port = sfccon.VxLAN_UDP_PORT
205 expect_udp_port = sfccon.VxLANGPE_UDP_PORT
207 dst_port = ether[UDP].dport
208 if dst_port != expect_udp_port:
209 raise RuntimeError("UDP dest port must be {0}, {1}".
210 format(expect_udp_port, dst_port))
212 payload_data = ether[Raw].load
214 if test_type == "Proxy Inbound":
215 VerifyPacket.check_vxlan_protocol(payload_data)
217 VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)