GPL/traffic_scripts/send_icmp_wait_for_reply.py

   1 #!/usr/bin/env python3
   2
   3 # Copyright (c) 2020 Cisco and/or its affiliates.
   4 #
   5 # SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
   6 #
   7 # Licensed under the Apache License 2.0 or
   8 # GNU General Public License v2.0 or later;  you may not use this file
   9 # except in compliance with one of these Licenses. You
  10 # may obtain a copy of the Licenses at:
  11 #
  12 #     http://www.apache.org/licenses/LICENSE-2.0
  13 #     https://www.gnu.org/licenses/old-licenses/gpl-2.0-standalone.html
  14 #
  15 # Note: If this file is linked with Scapy, which is GPLv2+, your use of it
  16 # must be under GPLv2+.  If at any point in the future it is no longer linked
  17 # with Scapy (or other GPLv2+ licensed software), you are free to choose Apache 2.
  18 #
  19 # Unless required by applicable law or agreed to in writing, software
  20 # distributed under the License is distributed on an "AS IS" BASIS,
  21 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  22 # See the License for the specific language governing permissions and
  23 # limitations under the License.
  24
  25 """Traffic script that sends an IP ICMPv4 or ICMPv6."""
  26
  27 import sys
  28 import ipaddress
  29
  30 from scapy.layers.inet import ICMP, IP
  31 from scapy.layers.inet6 import IPv6, ICMPv6EchoRequest, ICMPv6EchoReply,\
  32     ICMPv6ND_NS
  33 from scapy.layers.l2 import Ether
  34 from scapy.packet import Raw
  35
  36 from .PacketVerifier import RxQueue, TxQueue
  37 from .TrafficScriptArg import TrafficScriptArg
  38
  39
  40 def valid_ipv4(ip):
  41     """Check if IP address has the correct IPv4 address format.
  42
  43     :param ip: IP address.
  44     :type ip: str
  45     :return: True in case of correct IPv4 address format,
  46              otherwise return False.
  47     :rtype: bool
  48     """
  49     try:
  50         ipaddress.IPv4Address(ip)
  51         return True
  52     except (AttributeError, ipaddress.AddressValueError):
  53         return False
  54
  55
  56 def valid_ipv6(ip):
  57     """Check if IP address has the correct IPv6 address format.
  58
  59     :param ip: IP address.
  60     :type ip: str
  61     :return: True in case of correct IPv6 address format,
  62              otherwise return False.
  63     :rtype: bool
  64     """
  65     try:
  66         ipaddress.IPv6Address(ip)
  67         return True
  68     except (AttributeError, ipaddress.AddressValueError):
  69         return False
  70
  71
  72 def main():
  73     """Send ICMP echo request and wait for ICMP echo reply. It ignores all other
  74     packets."""
  75     args = TrafficScriptArg(
  76         [u"dst_mac", u"src_mac", u"dst_ip", u"src_ip", u"timeout"]
  77     )
  78
  79     dst_mac = args.get_arg(u"dst_mac")
  80     src_mac = args.get_arg(u"src_mac")
  81     dst_ip = args.get_arg(u"dst_ip")
  82     src_ip = args.get_arg(u"src_ip")
  83     tx_if = args.get_arg(u"tx_if")
  84     rx_if = args.get_arg(u"rx_if")
  85     timeout = int(args.get_arg(u"timeout"))
  86     wait_step = 1
  87
  88     rxq = RxQueue(rx_if)
  89     txq = TxQueue(tx_if)
  90     sent_packets = []
  91
  92     # Create empty ip ICMP packet
  93     if valid_ipv4(src_ip) and valid_ipv4(dst_ip):
  94         ip_layer = IP
  95         icmp_req = ICMP
  96         icmp_resp = ICMP
  97         icmp_type = 0  # echo-reply
  98     elif valid_ipv6(src_ip) and valid_ipv6(dst_ip):
  99         ip_layer = IP
 100         icmp_req = ICMPv6EchoRequest
 101         icmp_resp = ICMPv6EchoReply
 102         icmp_type = 0  # Echo Reply
 103     else:
 104         raise ValueError(u"IP not in correct format")
 105
 106     icmp_request = (
 107             Ether(src=src_mac, dst=dst_mac) /
 108             ip_layer(src=src_ip, dst=dst_ip) /
 109             icmp_req()
 110     )
 111
 112     # Send created packet on the interface
 113     icmp_request /= Raw()
 114     sent_packets.append(icmp_request)
 115     txq.send(icmp_request)
 116
 117     for _ in range(1000):
 118         while True:
 119             icmp_reply = rxq.recv(wait_step, ignore=sent_packets)
 120             if icmp_reply is None:
 121                 timeout -= wait_step
 122                 if timeout < 0:
 123                     raise RuntimeError(u"ICMP echo Rx timeout")
 124
 125             elif icmp_reply.haslayer(ICMPv6ND_NS):
 126                 # read another packet in the queue in case of ICMPv6ND_NS packet
 127                 continue
 128             else:
 129                 # otherwise process the current packet
 130                 break
 131
 132         if icmp_reply[ip_layer][icmp_resp].type == icmp_type:
 133             if icmp_reply[ip_layer].src == dst_ip and \
 134                     icmp_reply[ip_layer].dst == src_ip:
 135                 break
 136     else:
 137         raise RuntimeError(u"Max packet count limit reached")
 138
 139     print(u"ICMP echo reply received.")
 140
 141     sys.exit(0)
 142
 143
 144 if __name__ == u"__main__":
 145     main()