resources/traffic_scripts/send_icmp_check_multipath.py

   1 #!/usr/bin/env python
   2 # Copyright (c) 2016 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 IP ICMPv4/ICMPv6 packets traffic
  16 and check if it is divided into two paths."""
  17
  18 import sys
  19 import ipaddress
  20
  21 from scapy.all import Ether
  22 from scapy.layers.inet import ICMP, IP
  23 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6ND_NS
  24
  25 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  26 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  27
  28
  29 def valid_ipv4(ip):
  30     try:
  31         ipaddress.IPv4Address(unicode(ip))
  32         return True
  33     except (AttributeError, ipaddress.AddressValueError):
  34         return False
  35
  36
  37 def valid_ipv6(ip):
  38     try:
  39         ipaddress.IPv6Address(unicode(ip))
  40         return True
  41     except (AttributeError, ipaddress.AddressValueError):
  42         return False
  43
  44
  45 def main():
  46     """Send 100 IP ICMP packets traffic and check if it is divided into
  47     two paths."""
  48     args = TrafficScriptArg(
  49         ['src_ip', 'dst_ip', 'tg_if1_mac', 'dut_if1_mac', 'dut_if2_mac',
  50          'path_1_mac', 'path_2_mac'])
  51
  52     src_ip = args.get_arg('src_ip')
  53     dst_ip = args.get_arg('dst_ip')
  54     tg_if1_mac = args.get_arg('tg_if1_mac')
  55     dut_if1_mac = args.get_arg('dut_if1_mac')
  56     dut_if2_mac = args.get_arg('dut_if2_mac')
  57     path_1_mac = args.get_arg('path_1_mac')
  58     path_2_mac = args.get_arg('path_2_mac')
  59     tx_if = args.get_arg('tx_if')
  60     rx_if = args.get_arg('rx_if')
  61     path_1_counter = 0
  62     path_2_counter = 0
  63
  64     rxq = RxQueue(rx_if)
  65     txq = TxQueue(tx_if)
  66     sent_packets = []
  67     ip_format = ''
  68     pkt_raw = ''
  69     separator = ''
  70
  71     if valid_ipv4(src_ip):
  72         separator = '.'
  73     elif valid_ipv6(src_ip):
  74         separator = ':'
  75     else:
  76         raise ValueError("Source address not in correct format")
  77
  78     src_ip_base = (src_ip.rsplit(separator, 1))[0] + separator
  79
  80     for i in range(1, 101):
  81         if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
  82             pkt_raw = (Ether(src=tg_if1_mac, dst=dut_if1_mac) /
  83                        IP(src=src_ip_base+str(i), dst=dst_ip) /
  84                        ICMP())
  85             ip_format = 'IP'
  86         elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
  87             pkt_raw = (Ether(src=tg_if1_mac, dst=dut_if1_mac) /
  88                        IPv6(src=src_ip_base+str(i), dst=dst_ip) /
  89                        ICMPv6EchoRequest())
  90             ip_format = 'IPv6'
  91         else:
  92             raise ValueError("IP not in correct format")
  93
  94         sent_packets.append(pkt_raw)
  95         txq.send(pkt_raw)
  96
  97         while True:
  98             ether = rxq.recv(2)
  99             if ether is None:
 100                 raise RuntimeError('ICMPv6 echo reply Rx timeout')
 101
 102             if ether.haslayer(ICMPv6ND_NS):
 103                 # read another packet in the queue in case of ICMPv6ND_NS packet
 104                 continue
 105             else:
 106                 # otherwise process the current packet
 107                 break
 108
 109         if ether is None:
 110             raise RuntimeError("ICMP echo Rx timeout")
 111         if not ether.haslayer(ip_format):
 112             raise RuntimeError("Not an IP packet received {0}".
 113                                format(ether.__repr__()))
 114
 115         if ether[Ether].src != dut_if2_mac:
 116             raise RuntimeError("Source MAC address error")
 117
 118         if ether[Ether].dst == path_1_mac:
 119             path_1_counter += 1
 120         elif ether[Ether].dst == path_2_mac:
 121             path_2_counter += 1
 122         else:
 123             raise RuntimeError("Destination MAC address error")
 124
 125     if (path_1_counter + path_2_counter) != 100:
 126         raise RuntimeError("Packet loss: recevied only {} packets of 100 ".
 127                            format(path_1_counter + path_2_counter))
 128
 129     if path_1_counter == 0:
 130         raise RuntimeError("Path 1 error!")
 131
 132     if path_2_counter == 0:
 133         raise RuntimeError("Path 2 error!")
 134
 135     print "Path_1 counter: {}".format(path_1_counter)
 136     print "Path_2 counter: {}".format(path_2_counter)
 137
 138     sys.exit(0)
 139
 140
 141 if __name__ == "__main__":
 142     main()