resources/traffic_scripts/ipv4_ping_ttl_check.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Cisco and/or its affiliates.
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at:
   7 #
   8 #     http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 from scapy.all import Ether, IP, ICMP
  17 from resources.libraries.python.PacketVerifier \
  18     import Interface, create_gratuitous_arp_request, auto_pad, checksum_equal
  19 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  20
  21
  22 def check_ttl(ttl_begin, ttl_end, ttl_diff):
  23     if ttl_begin != ttl_end + ttl_diff:
  24         raise RuntimeError(
  25             "TTL changed from {} to {} but decrease by {} expected"
  26             .format(ttl_begin, ttl_end, ttl_diff))
  27
  28
  29 def ckeck_packets_equal(pkt_send, pkt_recv):
  30     pkt_send_raw = auto_pad(pkt_send)
  31     pkt_recv_raw = auto_pad(pkt_recv)
  32     if pkt_send_raw != pkt_recv_raw:
  33         print "Sent:     {}".format(pkt_send_raw.encode('hex'))
  34         print "Received: {}".format(pkt_recv_raw.encode('hex'))
  35         print "Sent:"
  36         pkt_send.show2()
  37         print "Received:"
  38         pkt_recv.show2()
  39         raise RuntimeError("Sent packet doesn't match received packet")
  40
  41
  42 def main():
  43     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
  44                              'hops', 'first_hop_mac', 'is_dst_tg'])
  45
  46     src_if_name = args.get_arg('tx_if')
  47     dst_if_name = args.get_arg('rx_if')
  48     is_dst_tg = True if args.get_arg('is_dst_tg') == 'True' else False
  49
  50     src_mac = args.get_arg('src_mac')
  51     first_hop_mac = args.get_arg('first_hop_mac')
  52     dst_mac = args.get_arg('dst_mac')
  53     src_ip = args.get_arg('src_ip')
  54     dst_ip = args.get_arg('dst_ip')
  55     hops = int(args.get_arg('hops'))
  56
  57     if is_dst_tg and (src_if_name == dst_if_name):
  58         raise RuntimeError(
  59             "Source interface name equals destination interface name")
  60
  61     src_if = Interface(src_if_name)
  62     src_if.send_pkt(str(create_gratuitous_arp_request(src_mac, src_ip)))
  63     if is_dst_tg:
  64         dst_if = Interface(dst_if_name)
  65         dst_if.send_pkt(str(create_gratuitous_arp_request(dst_mac, dst_ip)))
  66
  67     pkt_req_send = (Ether(src=src_mac, dst=first_hop_mac) /
  68                     IP(src=src_ip, dst=dst_ip) /
  69                     ICMP())
  70     src_if.send_pkt(pkt_req_send)
  71
  72     if is_dst_tg:
  73         pkt_req_recv = dst_if.recv_pkt()
  74         if pkt_req_recv is None:
  75             raise RuntimeError('Timeout waiting for packet')
  76
  77         check_ttl(pkt_req_send[IP].ttl, pkt_req_recv[IP].ttl, hops)
  78         pkt_req_send_mod = pkt_req_send.copy()
  79         pkt_req_send_mod[IP].ttl = pkt_req_recv[IP].ttl
  80         del pkt_req_send_mod[IP].chksum  # update checksum
  81         ckeck_packets_equal(pkt_req_send_mod[IP], pkt_req_recv[IP])
  82
  83         pkt_resp_send = (Ether(src=dst_mac, dst=pkt_req_recv.src) /
  84                          IP(src=dst_ip, dst=src_ip) /
  85                          ICMP(type=0))  # echo-reply
  86         dst_if.send_pkt(pkt_resp_send)
  87
  88     pkt_resp_recv = src_if.recv_pkt()
  89     if pkt_resp_recv is None:
  90         raise RuntimeError('Timeout waiting for packet')
  91
  92     if is_dst_tg:
  93         check_ttl(pkt_resp_send[IP].ttl, pkt_resp_recv[IP].ttl, hops)
  94         pkt_resp_send_mod = pkt_resp_send.copy()
  95         pkt_resp_send_mod[IP].ttl = pkt_resp_recv[IP].ttl
  96         del pkt_resp_send_mod[IP].chksum  # update checksum
  97         ckeck_packets_equal(pkt_resp_send_mod[IP], pkt_resp_recv[IP])
  98
  99     if not pkt_resp_recv.haslayer(IP):
 100         raise RuntimeError('Received packet does not contain IPv4 header: {}'.
 101                            format(pkt_resp_recv.__repr__()))
 102
 103     if pkt_resp_recv[IP].src != pkt_req_send[IP].dst:
 104         raise RuntimeError(
 105             'Received IPv4 packet contains wrong src IP address, '
 106             '{} instead of {}'.format(pkt_resp_recv[IP].src,
 107                                       pkt_req_send[IP].dst))
 108
 109     if pkt_resp_recv[IP].dst != pkt_req_send[IP].src:
 110         raise RuntimeError(
 111             'Received IPv4 packet contains wrong dst IP address, '
 112             '{} instead of {}'.format(pkt_resp_recv[IP].dst,
 113                                       pkt_req_send[IP].src))
 114
 115     # verify IPv4 checksum
 116     copy = pkt_resp_recv.copy()
 117     chksum = copy[IP].chksum
 118     del copy[IP].chksum
 119     tmp = IP(str(copy[IP]))
 120     if not checksum_equal(tmp.chksum, chksum):
 121         raise RuntimeError('Received IPv4 packet contains invalid checksum, '
 122                            '{} instead of {}'.format(chksum, tmp.chksum))
 123
 124     if not pkt_resp_recv[IP].haslayer(ICMP):
 125         raise RuntimeError(
 126             'Received IPv4 packet does not contain ICMP header: {}'.
 127             format(pkt_resp_recv[IP].__repr__()))
 128
 129     # verify ICMP checksum
 130     copy = pkt_resp_recv.copy()
 131     chksum = copy[IP][ICMP].chksum
 132     del copy[IP][ICMP].chksum
 133     tmp = ICMP(str(copy[IP][ICMP]))
 134     if not checksum_equal(tmp.chksum, chksum):
 135         raise RuntimeError('Received ICMP packet contains invalid checksum, '
 136                            '{} instead of {}'.format(chksum, tmp.chksum))
 137
 138     pkt_req_send_mod = pkt_req_send.copy()
 139     pkt_req_send_mod[IP][ICMP].type = pkt_resp_recv[IP][ICMP].type
 140     del pkt_req_send_mod[IP][ICMP].chksum  # update checksum
 141     ckeck_packets_equal(pkt_req_send_mod[IP][ICMP], pkt_resp_recv[IP][ICMP])
 142
 143
 144 if __name__ == "__main__":
 145     main()