resources/traffic_scripts/arp_request.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 """Send an ARP request and verify the reply"""
  17
  18 import sys
  19
  20 from scapy.all import Ether, ARP
  21
  22 from resources.libraries.python.PacketVerifier import Interface
  23 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  24
  25
  26 def parse_arguments():
  27     """Parse arguments of the script passed through command line
  28
  29     :return: tuple of parsed arguments
  30     """
  31     args = TrafficScriptArg(['src_if', 'src_mac', 'dst_mac',
  32                              'src_ip', 'dst_ip'])
  33
  34     # check for mandatory parameters
  35     params = (args.get_arg('tx_if'),
  36               args.get_arg('src_mac'),
  37               args.get_arg('dst_mac'),
  38               args.get_arg('src_ip'),
  39               args.get_arg('dst_ip'))
  40     if None in params:
  41         raise Exception('Missing mandatory parameter(s)!')
  42
  43     return params
  44
  45
  46 def arp_request_test():
  47     """Send ARP request, expect a reply and verify its fields.
  48
  49     returns: test status
  50     :raises RuntimeError: ARP reply timeout.
  51     """
  52     test_passed = False
  53     (src_if, src_mac, dst_mac, src_ip, dst_ip) = parse_arguments()
  54
  55     interface = Interface(src_if)
  56
  57     # build an ARP request
  58     arp_request = (Ether(src=src_mac, dst='ff:ff:ff:ff:ff:ff') /
  59                    ARP(psrc=src_ip, hwsrc=src_mac, pdst=dst_ip,
  60                        hwdst='ff:ff:ff:ff:ff:ff'))
  61
  62     # send the request
  63     interface.send_pkt(arp_request)
  64
  65     try:
  66         # wait for APR reply
  67         ether = interface.recv_pkt()
  68
  69         if not ether:
  70             raise RuntimeError("ARP reply timeout")
  71
  72         # verify received packet
  73
  74         if not ether.haslayer(ARP):
  75             raise RuntimeError('Unexpected packet: does not contain ARP ' +
  76                                'header "{}"'.format(ether.__repr__()))
  77
  78         arp = ether['ARP']
  79         arp_reply = 2
  80
  81         if arp.op != arp_reply:
  82             raise RuntimeError('expected op={}, received {}'.format(arp_reply,
  83                                                                     arp.op))
  84         if arp.ptype != 0x800:
  85             raise RuntimeError('expected ptype=0x800, received {}'.
  86                                format(arp.ptype))
  87         if arp.hwlen != 6:
  88             raise RuntimeError('expected hwlen=6, received {}'.
  89                                format(arp.hwlen))
  90         if arp.plen != 4:
  91             raise RuntimeError('expected plen=4, received {}'.format(arp.plen))
  92         if arp.hwsrc != dst_mac:
  93             raise RuntimeError('expected hwsrc={}, received {}'.
  94                                format(dst_mac, arp.hwsrc))
  95         if arp.psrc != dst_ip:
  96             raise RuntimeError('expected psrc={}, received {}'.
  97                                format(dst_ip, arp.psrc))
  98         if arp.hwdst != src_mac:
  99             raise RuntimeError('expected hwdst={}, received {}'.
 100                                format(src_mac, arp.hwdst))
 101         if arp.pdst != src_ip:
 102             raise RuntimeError('expected pdst={}, received {}'.
 103                                format(src_ip, arp.pdst))
 104         test_passed = True
 105
 106     except RuntimeError as ex:
 107         print 'Error occurred: {}'.format(ex)
 108
 109     return test_passed
 110
 111
 112 def main():
 113     """Run the test and collect result"""
 114     if arp_request_test():
 115         sys.exit(0)
 116     else:
 117         sys.exit(1)
 118
 119 if __name__ == '__main__':
 120     main()