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
69 :raises RuntimeError: If the vxlan protocol field verify fails.
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))
77 if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
78 raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
81 def check_vxlangpe_nsh_protocol(payload_data, test_type):
83 verify the vxlangpe and nsh protocol in the payload data.
85 :param payload_data: the payload data in the packet.
86 :param test_type: the functional test type.
87 :type payload_data: str
89 :raises RuntimeError: If the vxlangpe and nsh protocol
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))
98 if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
99 raise RuntimeError("next protocol not the NSH")
101 if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
102 raise RuntimeError("Unexpected VNI flag: {0}".
103 format(vxlangpe_pkt.vni))
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))
111 if nsh_pkt.OAM != 0 and nsh_pkt.OAM != 1:
112 raise RuntimeError("Unexpected NSH OAM: {0}".
115 if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
116 raise RuntimeError("NSH length {0} incorrect".
117 format(nsh_pkt.length))
119 if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
120 raise RuntimeError("NSH MD-Type {0} incorrect".
121 format(nsh_pkt.MDtype))
123 if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
124 raise RuntimeError("NSH next protocol {0} incorrect".
125 format(nsh_pkt.nextproto))
127 if test_type == "Proxy Outbound" or test_type == "SFF":
128 expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
130 expect_nsi = sfccon.NSH_DEFAULT_NSI
132 nsp_nsi = nsh_pkt.nsp_nsi
134 nsi = nsp_nsi & 0x000000FF
135 if nsp != sfccon.NSH_DEFAULT_NSP:
136 raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
138 if nsi != expect_nsi:
139 raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
142 if nsh_c1 != sfccon.NSH_DEFAULT_C1:
143 raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
146 if nsh_c2 != sfccon.NSH_DEFAULT_C2:
147 raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
150 if nsh_c3 != sfccon.NSH_DEFAULT_C3:
151 raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
154 if nsh_c4 != sfccon.NSH_DEFAULT_C4:
155 raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
159 def check_the_nsh_sfc_packet(ether, frame_size, test_type):
161 verify the NSH SFC functional test loopback packet field
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).
169 :type ether: scapy.Ether
170 :type frame_size: Integer
172 :raises RuntimeError: If the packet field verify fails.
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
183 expect_pkt_len = origin_size - 4
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,
191 if not ether.haslayer(IP):
192 raise RuntimeError("Not a IPv4 packet")
194 pkt_proto = ether[IP].proto
195 if pkt_proto != sfccon.UDP_PROTOCOL:
196 raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
198 if test_type == "Proxy Inbound":
199 expect_udp_port = sfccon.VxLAN_UDP_PORT
201 expect_udp_port = sfccon.VxLANGPE_UDP_PORT
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))
208 payload_data = ether[Raw].load
210 if test_type == "Proxy Inbound":
211 VerifyPacket.check_vxlan_protocol(payload_data)
213 VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)