1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
#!/usr/bin/env python
# Copyright (c) 2017 Cisco and/or its affiliates.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at:
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module defines the common functions.
"""
import ipaddress
from scapy.layers.inet import IP, UDP
from scapy.all import Raw
from resources.libraries.python.constants import Constants as con
from resources.libraries.python.SFC.SFCConstants import SFCConstants as sfccon
from resources.libraries.python.SFC.TunnelProtocol import VxLAN, VxLANGPE, NSH
from robot.api import logger
def valid_ipv4(ipaddr):
"""Check if IP address has the correct IPv4 address format.
:param ipaddr: IP address.
:type ipaddr: str
:returns: True in case of correct IPv4 address format,
otherwise return False.
:rtype: bool
"""
try:
ipaddress.IPv4Address(unicode(ipaddr))
return True
except (AttributeError, ipaddress.AddressValueError):
return False
def valid_ipv6(ipaddr):
"""Check if IP address has the correct IPv6 address format.
:param ipaddr: IP address.
:type ipaddr: str
:returns: True in case of correct IPv6 address format,
otherwise return False.
:rtype: bool
"""
try:
ipaddress.IPv6Address(unicode(ipaddr))
return True
except (AttributeError, ipaddress.AddressValueError):
return False
class VerifyPacket(object):
"""Define some functions for the test filed verify."""
@staticmethod
def check_vxlan_protocol(payload_data):
"""
verify the vxlan protocol in the payload data.
:param payload_data: the payload data in the packet.
:type payload_data: str
:returns: none
:raises RuntimeError: If the vxlan protocol field verify fails.
"""
# get the vxlan packet and check it
vxlan_pkt = VxLAN(payload_data[0:8])
if vxlan_pkt.flags != sfccon.VxLAN_FLAGS:
raise RuntimeError("Unexpected Vxlan flags: {0}".
format(vxlan_pkt.flags))
if vxlan_pkt.vni != sfccon.VxLAN_DEFAULT_VNI:
raise RuntimeError("Unexpected VNI flag: {0}".format(vxlan_pkt.vni))
@staticmethod
def check_vxlangpe_nsh_protocol(payload_data, test_type):
"""
verify the vxlangpe and nsh protocol in the payload data.
:param payload_data: the payload data in the packet.
:param test_type: the functional test type.
:type payload_data: str
:type test_type: str
:returns: none
:raises RuntimeError: If the vxlangpe and nsh protocol
field verify fails.
"""
# get the vxlan-gpe packet and check it
vxlangpe_pkt = VxLANGPE(payload_data[0:8])
if vxlangpe_pkt.flags != sfccon.VxLANGPE_FLAGS:
raise RuntimeError("Unexpected Vxlan-GPE flags: {0}".
format(vxlangpe_pkt.flags))
if vxlangpe_pkt.nextproto != sfccon.VxLANGPE_NEXT_PROTOCOL:
raise RuntimeError("next protocol not the NSH")
if vxlangpe_pkt.vni != sfccon.VxLANGPE_DEFAULT_VNI:
raise RuntimeError("Unexpected VNI flag: {0}".
format(vxlangpe_pkt.vni))
# get the NSH packet and check it
nsh_pkt = NSH(payload_data[8:32])
if nsh_pkt.flags != sfccon.NSH_FLAGS:
raise RuntimeError("Unexpected NSH flags: {0}".
format(nsh_pkt.flags))
if nsh_pkt.length != sfccon.NSH_HEADER_LENGTH:
raise RuntimeError("NSH length {0} incorrect".
format(nsh_pkt.length))
if nsh_pkt.MDtype != sfccon.NSH_DEFAULT_MDTYPE:
raise RuntimeError("NSH MD-Type {0} incorrect".
format(nsh_pkt.MDtype))
if nsh_pkt.nextproto != sfccon.NSH_NEXT_PROTOCOL:
raise RuntimeError("NSH next protocol {0} incorrect".
format(nsh_pkt.nextproto))
if test_type == "Proxy Outbound" or test_type == "SFF":
expect_nsi = sfccon.NSH_DEFAULT_NSI - 1
else:
expect_nsi = sfccon.NSH_DEFAULT_NSI
nsp_nsi = nsh_pkt.nsp_nsi
nsp = nsp_nsi >> 8
nsi = nsp_nsi & 0x000000FF
if nsp != sfccon.NSH_DEFAULT_NSP:
raise RuntimeError("NSH Service Path ID {0} incorrect".format(nsp))
if nsi != expect_nsi:
raise RuntimeError("NSH Service Index {0} incorrect".format(nsi))
nsh_c1 = nsh_pkt.c1
if nsh_c1 != sfccon.NSH_DEFAULT_C1:
raise RuntimeError("NSH c1 {0} incorrect".format(nsh_c1))
nsh_c2 = nsh_pkt.c2
if nsh_c2 != sfccon.NSH_DEFAULT_C2:
raise RuntimeError("NSH c2 {0} incorrect".format(nsh_c2))
nsh_c3 = nsh_pkt.c3
if nsh_c3 != sfccon.NSH_DEFAULT_C3:
raise RuntimeError("NSH c3 {0} incorrect".format(nsh_c3))
nsh_c4 = nsh_pkt.c4
if nsh_c4 != sfccon.NSH_DEFAULT_C4:
raise RuntimeError("NSH c4 {0} incorrect".format(nsh_c4))
@staticmethod
def check_the_nsh_sfc_packet(ether, frame_size, test_type):
"""
verify the NSH SFC functional test loopback packet field
is correct.
:param ether: The Ethernet packet data.
:param frame_size: The origin frame size.
:param test_type: The test type.
(Classifier, Proxy Inbound, Proxy Outbound, SFF).
:type ether: scapy.Ether
:type frame_size: Integer
:type test_type: str
:returns: none
:raises RuntimeError: If the packet field verify fails.
"""
origin_size = int(frame_size)
if test_type == "Classifier":
expect_pkt_len = origin_size + 74 - 4
elif test_type == "Proxy Inbound":
expect_pkt_len = origin_size - 24 - 4
elif test_type == "Proxy Outbound":
expect_pkt_len = origin_size + 24 - 4
else:
expect_pkt_len = origin_size - 4
recv_pkt_len = len(ether)
if recv_pkt_len != expect_pkt_len:
raise RuntimeError("Received packet size {0} not " \
"the expect size {1}".format(recv_pkt_len, \
expect_pkt_len))
if not ether.haslayer(IP):
raise RuntimeError("Not a IPv4 packet")
pkt_proto = ether[IP].proto
if pkt_proto != sfccon.UDP_PROTOCOL:
raise RuntimeError("Not a UDP packet , {0}".format(pkt_proto))
if test_type == "Proxy Inbound":
expect_udp_port = sfccon.VxLAN_UDP_PORT
else:
expect_udp_port = sfccon.VxLANGPE_UDP_PORT
dst_port = ether[UDP].dport
if dst_port != expect_udp_port:
raise RuntimeError("UDP dest port must be {0}, {1}".
format(expect_udp_port, dst_port))
payload_data = ether[Raw].load
if test_type == "Proxy Inbound":
VerifyPacket.check_vxlan_protocol(payload_data)
else:
VerifyPacket.check_vxlangpe_nsh_protocol(payload_data, test_type)
|
