resources/traffic_scripts/send_vxlan_for_proxy_test.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 """Traffic script that sends an VxLAN packet
  16 from TG to DUT.
  17 """
  18 import sys
  19 import time
  20
  21 from scapy.layers.inet import IP, UDP, TCP
  22 from scapy.layers.inet6 import IPv6, ICMPv6ND_NS
  23 from scapy.all import Ether, Packet, Raw
  24
  25 from resources.libraries.python.SFC.VerifyPacket import *
  26 from resources.libraries.python.SFC.SFCConstants import SFCConstants as SfcCon
  27 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  28 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  29
  30
  31 def main():
  32     """Send VxLAN packet from TG to DUT.
  33
  34     :raises: If the IP address is invalid.
  35     """
  36     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
  37                              'timeout', 'framesize', 'testtype'])
  38
  39     src_mac = args.get_arg('src_mac')
  40     dst_mac = args.get_arg('dst_mac')
  41     src_ip = args.get_arg('src_ip')
  42     dst_ip = args.get_arg('dst_ip')
  43     tx_if = args.get_arg('tx_if')
  44     rx_if = args.get_arg('rx_if')
  45     timeout = max(2, int(args.get_arg('timeout')))
  46     frame_size = int(args.get_arg('framesize'))
  47     test_type = args.get_arg('testtype')
  48
  49     rxq = RxQueue(rx_if)
  50     txq = TxQueue(tx_if)
  51     sent_packets = []
  52
  53     protocol = TCP
  54     source_port = SfcCon.DEF_SRC_PORT
  55     destination_port = SfcCon.DEF_DST_PORT
  56
  57     if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
  58         ip_version = IP
  59     elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
  60         ip_version = IPv6
  61     else:
  62         raise ValueError("Invalid IP version!")
  63
  64     innerpkt = (Ether(src=src_mac, dst=dst_mac) /
  65                 ip_version(src=src_ip, dst=dst_ip) /
  66                 protocol(sport=int(source_port), dport=int(destination_port)))
  67
  68     vxlan = '\x08\x00\x00\x00\x00\x00\x01\x00'
  69
  70     raw_data = vxlan + str(innerpkt)
  71
  72     pkt_header = (Ether(src=src_mac, dst=dst_mac) /
  73                   ip_version(src=src_ip, dst=dst_ip) /
  74                   UDP(sport=int(source_port), dport=4789) /
  75                   Raw(load=raw_data))
  76
  77     fsize_no_fcs = frame_size - 4
  78     pad_len = max(0, fsize_no_fcs - len(pkt_header))
  79     pad_data = "A" * pad_len
  80
  81     pkt_raw = pkt_header / Raw(load=pad_data)
  82
  83     # Send created packet on one interface and receive on the other
  84     sent_packets.append(pkt_raw)
  85     txq.send(pkt_raw)
  86
  87     while True:
  88         ether = rxq.recv(timeout)
  89         if ether is None:
  90             raise RuntimeError('No packet is received!')
  91
  92         if ether.haslayer(ICMPv6ND_NS):
  93             # read another packet in the queue if the current one is ICMPv6ND_NS
  94             continue
  95         else:
  96             # otherwise process the current packet
  97             break
  98
  99     # let us begin to check the proxy outbound packet
 100     VerifyPacket.check_the_nsh_sfc_packet(ether, frame_size, test_type)
 101
 102     # we check all the fields about the proxy outbound, this test will pass
 103     sys.exit(0)
 104
 105
 106 if __name__ == "__main__":
 107     main()