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.
15 """Traffic script that sends an VxLAN packet
21 from scapy.layers.inet import IP, UDP, TCP
22 from scapy.layers.inet6 import IPv6
23 from scapy.all import Ether, Packet, Raw
24 from scapy.all import sendp
26 from resources.libraries.python.SFC.VerifyPacket import *
27 from resources.libraries.python.SFC.SFCConstants import SFCConstants as sfccon
28 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
30 from robot.api import logger
33 """Send VxLAN packet from TG to DUT.
35 :raises: If the IP address is invalid.
37 args = TrafficScriptArg(
38 ['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
39 'timeout', 'framesize', 'testtype'])
41 src_mac = args.get_arg('src_mac')
42 dst_mac = args.get_arg('dst_mac')
43 src_ip = args.get_arg('src_ip')
44 dst_ip = args.get_arg('dst_ip')
45 tx_if = args.get_arg('tx_if')
46 rx_if = args.get_arg('rx_if')
47 timeout = int(args.get_arg('timeout'))
48 frame_size = int(args.get_arg('framesize'))
49 test_type = args.get_arg('testtype')
52 source_port = sfccon.DEF_SRC_PORT
53 destination_port = sfccon.DEF_DST_PORT
56 if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
58 elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
61 raise ValueError("Invalid IP version!")
63 innerpkt = (Ether(src=src_mac, dst=dst_mac) /
64 ip_version(src=src_ip, dst=dst_ip) /
65 protocol(sport=int(source_port), dport=int(destination_port)))
67 vxlan = '\x08\x00\x00\x00\x00\x00\x01\x00'
69 raw_data = vxlan + str(innerpkt)
71 pkt_header = (Ether(src=src_mac, dst=dst_mac) /
72 ip_version(src=src_ip, dst=dst_ip) /
73 UDP(sport=int(source_port), dport=4789) /
76 fsize_no_fcs = frame_size - 4
77 pad_len = max(0, fsize_no_fcs - len(pkt_header))
78 pad_data = "A" * pad_len
80 pkt_raw = pkt_header / Raw(load=pad_data)
82 sendp(pkt_raw, iface=tx_if, count=3)
86 # let us begin to check the proxy outbound packet
87 VerifyPacket.check_the_nsh_sfc_packet(frame_size, test_type)
89 # we check all the fields about the proxy outbound, this test will pass
93 if __name__ == "__main__":