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 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
28 from robot.api import logger
30 def valid_ipv4(ipaddr):
31 """Check if IP address has the correct IPv4 address format.
33 :param ipaddr: IP address.
35 :returns: True in case of correct IPv4 address format,
36 otherwise return False.
40 ipaddress.IPv4Address(unicode(ipaddr))
42 except (AttributeError, ipaddress.AddressValueError):
46 def valid_ipv6(ipaddr):
47 """Check if IP address has the correct IPv6 address format.
49 :param ipaddr: IP address.
51 :returns: True in case of correct IPv6 address format,
52 otherwise return False.
56 ipaddress.IPv6Address(unicode(ipaddr))
58 except (AttributeError, ipaddress.AddressValueError):
62 class VerifyPacket(object):
63 """Define some functions for the test filed verify."""
66 def check_vxlan_protocol(payload_data):
68 verify the vxlan protocol in the payload data.
70 :param payload_data: the payload data in the packet.
71 :type payload_data: str
73 :raises RuntimeError: If the vxlan protocol field verify fails.
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))
81 if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
82 raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
85 def check_vxlangpe_nsh_protocol(payload_data, test_type):
87 verify the vxlangpe and nsh protocol in the payload data.
89 :param payload_data: the payload data in the packet.
90 :param test_type: the functional test type.
91 :type payload_data: str
94 :raises RuntimeError: If the vxlangpe and nsh protocol
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))
103 if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
104 raise RuntimeError("next protocol not the NSH")
106 if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
107 raise RuntimeError("Unexpected VNI flag: {0}".
108 format(vxlangpe_pkt.vni))
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))
116 if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
117 raise RuntimeError("NSH length {0} incorrect".
118 format(nsh_pkt.length))
120 if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
121 raise RuntimeError("NSH MD-Type {0} incorrect".
122 format(nsh_pkt.MDtype))
124 if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
125 raise RuntimeError("NSH next protocol {0} incorrect".
126 format(nsh_pkt.nextproto))
128 if test_type == "Proxy Outbound" or test_type == "SFF":
129 expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
131 expect_nsi = sfccon.NSH_DEFAULT_NSI
133 nsp_nsi = nsh_pkt.nsp_nsi
135 nsi = nsp_nsi & 0x000000FF
136 if nsp != sfccon.NSH_DEFAULT_NSP:
137 raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
139 if nsi != expect_nsi:
140 raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
143 if nsh_c1 != sfccon.NSH_DEFAULT_C1:
144 raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
147 if nsh_c2 != sfccon.NSH_DEFAULT_C2:
148 raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
151 if nsh_c3 != sfccon.NSH_DEFAULT_C3:
152 raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
155 if nsh_c4 != sfccon.NSH_DEFAULT_C4:
156 raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
160 def check_the_nsh_sfc_packet(frame_size, test_type):
162 verify the NSH SFC functional test loopback packet field
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
171 :raises RuntimeError: If the packet field verify fails.
174 rx_pcapfile = '{0}/nsh_sfc_tests/sfc_scripts/temp_packet.pcap' \
175 .format(con.REMOTE_FW_DIR)
177 logger.trace('read pcap file:{0}'.format(rx_pcapfile))
179 packets = rdpcap(rx_pcapfile)
181 raise RuntimeError("No packet is received!")
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
193 expect_pkt_len = origin_size - 4
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, \
201 if not ether.haslayer(IP):
202 raise RuntimeError("Not a IPv4 packet")
204 pkt_proto = ether[IP].proto
205 if pkt_proto != sfccon.UDP_PROTOCOL:
206 raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
208 if test_type == "Proxy Inbound":
209 expect_udp_port = sfccon.VxLAN_UDP_PORT
211 expect_udp_port = sfccon.VxLANGPE_UDP_PORT
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))
218 payload_data = ether[Raw].load
220 if test_type == "Proxy Inbound":
221 VerifyPacket.check_vxlan_protocol(payload_data)
223 VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)