resources/traffic_scripts/ipv6_sweep_ping.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 """Traffic script for IPv6 sweep ping."""
  17
  18 import logging
  19 import os
  20 import sys
  21
  22 # pylint: disable=no-name-in-module
  23 # pylint: disable=import-error
  24 logging.getLogger("scapy.runtime").setLevel(logging.ERROR)
  25
  26 from scapy.all import Ether
  27 from scapy.layers.inet6 import IPv6, ICMPv6ND_NA, ICMPv6ND_NS
  28 from scapy.layers.inet6 import ICMPv6NDOptDstLLAddr
  29 from scapy.layers.inet6 import ICMPv6EchoRequest, ICMPv6EchoReply
  30
  31 from resources.libraries.python.PacketVerifier import RxQueue, TxQueue
  32 from resources.libraries.python.PacketVerifier import checksum_equal
  33 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  34
  35
  36 def main():
  37     # start_size - start size of the ICMPv6 echo data
  38     # end_size - end size of the ICMPv6 echo data
  39     # step - increment step
  40     args = TrafficScriptArg(['src_mac', 'dst_mac', 'src_ip', 'dst_ip',
  41                              'start_size', 'end_size', 'step'])
  42
  43     rxq = RxQueue(args.get_arg('rx_if'))
  44     txq = TxQueue(args.get_arg('tx_if'))
  45
  46     src_mac = args.get_arg('src_mac')
  47     dst_mac = args.get_arg('dst_mac')
  48     src_ip = args.get_arg('src_ip')
  49     dst_ip = args.get_arg('dst_ip')
  50     start_size = int(args.get_arg('start_size'))
  51     end_size = int(args.get_arg('end_size'))
  52     step = int(args.get_arg('step'))
  53     echo_id = 0xa
  54     # generate some random data buffer
  55     data = bytearray(os.urandom(end_size))
  56
  57     # send ICMPv6 neighbor advertisement message
  58     sent_packets = []
  59     pkt_send = (Ether(src=src_mac, dst=dst_mac) /
  60                 IPv6(src=src_ip, dst=dst_ip) /
  61                 ICMPv6ND_NA(tgt=src_ip, R=0) /
  62                 ICMPv6NDOptDstLLAddr(lladdr=src_mac))
  63     sent_packets.append(pkt_send)
  64     txq.send(pkt_send)
  65
  66     # send ICMPv6 echo request with incremented data length and receive ICMPv6
  67     # echo reply
  68     for echo_seq in range(start_size, end_size + 1, step):
  69         pkt_send = (Ether(src=src_mac, dst=dst_mac) /
  70                     IPv6(src=src_ip, dst=dst_ip) /
  71                     ICMPv6EchoRequest(id=echo_id, seq=echo_seq,
  72                                       data=data[0:echo_seq]))
  73         sent_packets.append(pkt_send)
  74         txq.send(pkt_send)
  75
  76         while True:
  77             ether = rxq.recv(ignore=sent_packets)
  78             if ether is None:
  79                 raise RuntimeError('ICMPv6 echo reply seq {0} '
  80                                    'Rx timeout'.format(echo_seq))
  81
  82             if ether.haslayer(ICMPv6ND_NS):
  83                 # read another packet in the queue in case of ICMPv6ND_NS packet
  84                 continue
  85             else:
  86                 # otherwise process the current packet
  87                 break
  88
  89         if not ether.haslayer(IPv6):
  90             raise RuntimeError('Unexpected packet with no IPv6 layer '
  91                                'received: {0}'.format(ether.__repr__()))
  92
  93         ipv6 = ether[IPv6]
  94
  95         if not ipv6.haslayer(ICMPv6EchoReply):
  96             raise RuntimeError('Unexpected packet with no ICMPv6 echo reply '
  97                                'layer received: {0}'.format(ipv6.__repr__()))
  98
  99         icmpv6 = ipv6[ICMPv6EchoReply]
 100
 101         if icmpv6.id != echo_id or icmpv6.seq != echo_seq:
 102             raise RuntimeError('ICMPv6 echo reply with invalid data received: '
 103                                'ID {0} seq {1} should be ID {2} seq {3}, {0}'.
 104                                format(icmpv6.id, icmpv6.seq, echo_id,
 105                                       echo_seq))
 106
 107         cksum = icmpv6.cksum
 108         del icmpv6.cksum
 109         tmp = ICMPv6EchoReply(str(icmpv6))
 110         if not checksum_equal(tmp.cksum, cksum):
 111             raise RuntimeError('Invalid checksum: {0} should be {1}'.
 112                                format(cksum, tmp.cksum))
 113
 114         sent_packets.remove(pkt_send)
 115
 116     sys.exit(0)
 117
 118
 119 if __name__ == "__main__":
 120     main()