resources/traffic_scripts/dhcp/check_dhcp_discover.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 receives an DHCP packet on given interface and check if
  16 is correct DHCP DISCOVER message.
  17 """
  18
  19 import sys
  20
  21 from scapy.layers.inet import UDP_SERVICES
  22
  23 from resources.libraries.python.PacketVerifier import RxQueue
  24 from resources.libraries.python.TrafficScriptArg import TrafficScriptArg
  25
  26
  27 def main():
  28     args = TrafficScriptArg(['rx_src_mac'], ['hostname'])
  29
  30     rx_if = args.get_arg('rx_if')
  31     rx_src_mac = args.get_arg('rx_src_mac')
  32     hostname = args.get_arg('hostname')
  33
  34     rx_dst_mac = 'ff:ff:ff:ff:ff:ff'
  35     rx_src_ip = '0.0.0.0'
  36     rx_dst_ip = '255.255.255.255'
  37     boot_request = 1
  38     dhcp_magic = 'c\x82Sc'
  39
  40     rxq = RxQueue(rx_if)
  41
  42     ether = rxq.recv(10)
  43
  44     if ether is None:
  45         raise RuntimeError("DHCP DISCOVER Rx timeout.")
  46
  47     if ether.dst != rx_dst_mac:
  48         raise RuntimeError("Destination MAC address error.")
  49     print "Destination MAC address: OK."
  50
  51     if ether.src != rx_src_mac:
  52         raise RuntimeError("Source MAC address error.")
  53     print "Source MAC address: OK."
  54
  55     if ether['IP'].dst != rx_dst_ip:
  56         raise RuntimeError("Destination IP address error.")
  57     print "Destination IP address: OK."
  58
  59     if ether['IP'].src != rx_src_ip:
  60         raise RuntimeError("Source IP address error.")
  61     print "Source IP address: OK."
  62
  63     if ether['IP']['UDP'].dport != UDP_SERVICES.bootps:
  64         raise RuntimeError("UDP destination port error.")
  65     print "UDP destination port: OK."
  66
  67     if ether['IP']['UDP'].sport != UDP_SERVICES.bootpc:
  68         raise RuntimeError("UDP source port error.")
  69     print "UDP source port: OK."
  70
  71     bootp = ether['BOOTP']
  72
  73     if bootp.op != boot_request:
  74         raise RuntimeError("BOOTP message type error.")
  75     print "BOOTP message type: OK"
  76
  77     if bootp.ciaddr != '0.0.0.0':
  78         raise RuntimeError("BOOTP client IP address error.")
  79     print "BOOTP client IP address: OK"
  80
  81     if bootp.yiaddr != '0.0.0.0':
  82         raise RuntimeError("BOOTP 'your' (client) IP address error.")
  83     print "BOOTP 'your' (client) IP address: OK"
  84
  85     if bootp.siaddr != '0.0.0.0':
  86         raise RuntimeError("BOOTP next server IP address error.")
  87     print "BOOTP next server IP address: OK"
  88
  89     if bootp.giaddr != '0.0.0.0':
  90         raise RuntimeError("BOOTP relay agent IP address error.")
  91     print "BOOTP relay agent IP address: OK"
  92
  93     chaddr = bootp.chaddr[:bootp.hlen].encode('hex')
  94     if chaddr != rx_src_mac.replace(':', ''):
  95         raise RuntimeError("BOOTP client hardware address error.")
  96     print "BOOTP client hardware address: OK"
  97
  98     # Check hostname
  99     if bootp.sname != 64*'\x00':
 100         raise RuntimeError("BOOTP server name error.")
 101     print "BOOTP server name: OK"
 102
 103     # Check boot file
 104     if bootp.file != 128*'\x00':
 105         raise RuntimeError("BOOTP boot file name error.")
 106     print "BOOTP boot file name: OK"
 107
 108     # Check bootp magic
 109     if bootp.options != dhcp_magic:
 110         raise RuntimeError("DHCP magic error.")
 111     print "DHCP magic: OK"
 112
 113     # Check options
 114     dhcp_options = ether['DHCP options'].options
 115
 116     # Option 12
 117     hn = filter(lambda x: x[0] == 'hostname', dhcp_options)
 118     if hostname:
 119         try:
 120             if hn[0][1] != hostname:
 121                 raise RuntimeError("Client's hostname doesn't match.")
 122         except IndexError:
 123             raise RuntimeError("Option list doesn't contain hostname option.")
 124     else:
 125         if len(hn) != 0:
 126             raise RuntimeError("Option list contains hostname option.")
 127     print "Option 12 hostname: OK"
 128
 129     # Option 53
 130     mt = filter(lambda x: x[0] == 'message-type', dhcp_options)[0][1]
 131     if mt != 1:
 132         raise RuntimeError("Option 53 message-type error.")
 133     print "Option 53 message-type: OK"
 134
 135     # Option 55
 136     prl = filter(lambda x: x[0] == 'param_req_list', dhcp_options)[0][1]
 137     if prl != '\x01\x1c\x02\x03\x0f\x06w\x0c,/\x1ay*':
 138         raise RuntimeError("Option 55 param_req_list error.")
 139     print "Option 55 param_req_list: OK"
 140
 141     # Option 255
 142     if 'end' not in dhcp_options:
 143         raise RuntimeError("end option error.")
 144     print "end option: OK"
 145
 146     sys.exit(0)
 147
 148
 149 if __name__ == "__main__":
 150     main()