resources/traffic_scripts/span_check.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 packet from one interface
  16 to the other. Source and destination IP addresses and source and destination
  17 MAC addresses are checked in received packet.
  18 """
  19
  20 import sys
  21 import ipaddress
  22
  23 from scapy.layers.inet import IP, ICMP, ARP
  24 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply
  25 from scapy.layers.l2 import Ether
  26
  27 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue, auto_pad
  28 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  29
  30
  31 def valid_ipv4(address):
  32     """Check if IP address has the correct IPv4 address format.
  33
  34     :param address: IP address.
  35     :type address: str
  36     :return: True in case of correct IPv4 address format,
  37     otherwise return false.
  38     :rtype: bool
  39     """
  40     try:
  41         ipaddress.IPv4Address(unicode(address))
  42         return True
  43     except (AttributeError, ipaddress.AddressValueError):
  44         return False
  45
  46
  47 def valid_ipv6(address):
  48     """Check if IP address has the correct IPv6 address format.
  49
  50     :param address: IP address.
  51     :type address: str
  52     :return: True in case of correct IPv6 address format,
  53     otherwise return false.
  54     :rtype: bool
  55     """
  56     try:
  57         ipaddress.IPv6Address(unicode(address))
  58         return True
  59     except (AttributeError, ipaddress.AddressValueError):
  60         return False
  61
  62
  63 def main():
  64     """Send a simple L2 or ICMP packet from one TG interface to DUT, then
  65     receive a copy of the packet on the second TG interface, and a copy of
  66     the ICMP reply."""
  67     args = TrafficScriptArg(
  68         ['tg_src_mac', 'src_ip', 'dst_ip', 'dut_if1_mac', 'ptype'])
  69
  70     src_mac = args.get_arg('tg_src_mac')
  71     dst_mac = args.get_arg('dut_if1_mac')
  72     src_ip = args.get_arg('src_ip')
  73     dst_ip = args.get_arg('dst_ip')
  74     tx_if = args.get_arg('tx_if')
  75     rx_if = args.get_arg('rx_if')
  76     ptype = args.get_arg('ptype')
  77
  78     rxq = RxQueue(rx_if)
  79     txq = TxQueue(tx_if)
  80
  81     if ptype == "ARP":
  82         pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
  83                    ARP(hwsrc=src_mac, hwdst="00:00:00:00:00:00",
  84                        psrc=src_ip, pdst=dst_ip, op="who-has"))
  85     elif ptype == "ICMP":
  86         if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
  87             pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
  88                        IP(src=src_ip, dst=dst_ip) /
  89                        ICMP(type="echo-request"))
  90         elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
  91             pkt_raw = (Ether(src=src_mac, dst=dst_mac) /
  92                        IPv6(src=src_ip, dst=dst_ip) /
  93                        ICMPv6EchoRequest())
  94         else:
  95             raise ValueError("IP not in correct format")
  96     else:
  97         raise RuntimeError("Unexpected payload type.")
  98
  99     txq.send(pkt_raw)
 100     ether = rxq.recv(2)
 101
 102     # Receive copy of sent packet.
 103     if ether is None:
 104         raise RuntimeError("Rx timeout")
 105     pkt = auto_pad(pkt_raw)
 106     if str(ether) != str(pkt):
 107         raise RuntimeError("Mirrored packet does not match packet sent.")
 108
 109     # Receive copy of reply to sent packet.
 110     ether = rxq.recv(2)
 111     if ether is None:
 112         raise RuntimeError("Rx timeout")
 113     if ether.src != dst_mac or ether.dst != src_mac:
 114         raise RuntimeError("MAC mismatch in mirrored response.")
 115     if ptype == "ARP":
 116         if ether['ARP'].op != 2:
 117             raise RuntimeError("Mirrored packet is not an ARP reply.")
 118         if ether['ARP'].hwsrc != dst_mac or ether['ARP'].hwdst != src_mac:
 119             raise RuntimeError("ARP MAC does not match l2 MAC "
 120                                "in mirrored response.")
 121         if ether['ARP'].psrc != dst_ip or ether['ARP'].pdst != src_ip:
 122             raise RuntimeError("ARP IP address mismatch in mirrored response.")
 123     elif ptype == "ICMP" and ether.haslayer(IP):
 124         if ether['IP'].src != dst_ip or ether['IP'].dst != src_ip:
 125             raise RuntimeError("IP address mismatch in mirrored reply.")
 126         if ether['ICMP'].type != 0:
 127             raise RuntimeError("Mirrored packet is not an ICMP reply.")
 128     elif ptype == "ICMP" and ether.haslayer(IPv6):
 129         if ether['IPv6'].src != dst_ip or ether['IPv6'].dst != src_ip:
 130             raise RuntimeError("IP address mismatch in mirrored reply.")
 131         if not ether.haslayer(ICMPv6EchoReply):
 132             raise RuntimeError("Mirrored packet is not an ICMP reply.")
 133
 134     sys.exit(0)
 135
 136
 137 if __name__ == "__main__":
 138     main()