84eb6aa0a247cb3592dc60042245827e213742f4
[csit.git] / 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.Version != 0:
110             raise RuntimeError("Unexpected NSH version: {0}".
111                                format(nsh_pkt.Version))
112
113         print type(nsh_pkt.OAM)
114         if nsh_pkt.OAM != 0 or nsh_pkt.OAM != 1:
115             raise RuntimeError("Unexpected NSH OAM: {0}".
116                                format(nsh_pkt.OAM))
117
118         if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
119             raise RuntimeError("NSH length {0} incorrect".
120                                format(nsh_pkt.length))
121
122         if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
123             raise RuntimeError("NSH MD-Type {0} incorrect".
124                                format(nsh_pkt.MDtype))
125
126         if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
127             raise RuntimeError("NSH next protocol {0} incorrect".
128                                format(nsh_pkt.nextproto))
129
130         if test_type == "Proxy Outbound" or test_type == "SFF":
131             expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
132         else:
133             expect_nsi = sfccon.NSH_DEFAULT_NSI
134
135         nsp_nsi = nsh_pkt.nsp_nsi
136         nsp = nsp_nsi >> 8
137         nsi = nsp_nsi & 0x000000FF
138         if nsp != sfccon.NSH_DEFAULT_NSP:
139             raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
140
141         if nsi != expect_nsi:
142             raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
143
144         nsh_c1 = nsh_pkt.c1
145         if nsh_c1 != sfccon.NSH_DEFAULT_C1:
146             raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
147
148         nsh_c2 = nsh_pkt.c2
149         if nsh_c2 != sfccon.NSH_DEFAULT_C2:
150             raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
151
152         nsh_c3 = nsh_pkt.c3
153         if nsh_c3 != sfccon.NSH_DEFAULT_C3:
154             raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
155
156         nsh_c4 = nsh_pkt.c4
157         if nsh_c4 != sfccon.NSH_DEFAULT_C4:
158             raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
159
160
161     @staticmethod
162     def check_the_nsh_sfc_packet(ether, frame_size, test_type):
163         """
164         verify the NSH SFC functional test loopback packet field
165         is correct.
166
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).
171
172         :type ether: scapy.Ether
173         :type frame_size: Integer
174         :type test_type: str
175         :returns: none
176         :raises RuntimeError: If the packet field verify fails.
177         """
178
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
186         else:
187             expect_pkt_len = origin_size - 4
188
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,
193                                                             expect_pkt_len))
194
195         if not ether.haslayer(IP):
196             raise RuntimeError("Not a IPv4 packet")
197
198         pkt_proto = ether[IP].proto
199         if pkt_proto != sfccon.UDP_PROTOCOL:
200             raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
201
202         if test_type == "Proxy Inbound":
203             expect_udp_port = sfccon.VxLAN_UDP_PORT
204         else:
205             expect_udp_port = sfccon.VxLANGPE_UDP_PORT
206
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))
211
212         payload_data = ether[Raw].load
213
214         if test_type == "Proxy Inbound":
215             VerifyPacket.check_vxlan_protocol(payload_data)
216         else:
217             VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)

©2016 FD.io a Linux Foundation Collaborative Project. All Rights Reserved.
Linux Foundation is a registered trademark of The Linux Foundation. Linux is a registered trademark of Linus Torvalds.
Please see our privacy policy and terms of use.