liminix

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README | LICENSE

test-dhcp-service.py (2358B)

      1 # forge packets for testing liminix and send them via the qemu udp
      2 # multicast socket interface
      3 
      4 MCAST_GRP = '230.0.0.1'
      5 MCAST_PORT = 1235
      6 MULTICAST_TTL = 2
      7 
      8 TIMEOUT = 10                    # seconds
      9 
     10 from warnings import filterwarnings
     11 filterwarnings("ignore")
     12 
     13 import random
     14 import binascii
     15 import socket
     16 import time
     17 import json
     18 
     19 from builtins import bytes, bytearray
     20 
     21 from scapy.all import Ether, IP, UDP, BOOTP, DHCP, sendp, send, raw
     22 
     23 class JSONEncoderWithBytes(json.JSONEncoder):
     24     def default(self, obj):
     25         if isinstance(obj, (bytes, bytearray)):
     26             return obj.decode('utf-8')
     27         return json.JSONEncoder.default(self, obj)
     28 
     29 
     30 def dhcp_option(pkt, label):
     31     if pkt.haslayer(DHCP):
     32         for i in pkt[DHCP].options:
     33             l, v = i
     34             if l == label:
     35                 return v
     36     return None
     37 
     38 def is_dhcp_offer(pkt):
     39     val = dhcp_option(pkt, 'message-type')
     40     return (val == 2)
     41 
     42 
     43 
     44 def mac_to_bytes(mac_addr: str) -> bytes:
     45     """ Converts a MAC address string to bytes.
     46     """
     47     return int(mac_addr.replace(":", ""), 16).to_bytes(6, "big")
     48 
     49 
     50 client_mac = "01:02:03:04:05:06"
     51 discover = (
     52     Ether(dst="ff:ff:ff:ff:ff:ff") /
     53     IP(src="0.0.0.0", dst="255.255.255.255") /
     54     UDP(sport=68, dport=67) /
     55     BOOTP(
     56         chaddr=mac_to_bytes(client_mac),
     57         xid=random.randint(1, 2**32-1),
     58     ) /
     59     DHCP(options=[("message-type", "discover"), "end"])
     60 )
     61 payload = raw(discover)
     62 
     63 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
     64 sock.settimeout(TIMEOUT)
     65 sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
     66 sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, MULTICAST_TTL)
     67 
     68 sock.bind((MCAST_GRP, MCAST_PORT))
     69 sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF, socket.inet_aton('127.0.0.1'))
     70 sock.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
     71                 socket.inet_aton(MCAST_GRP) + socket.inet_aton('127.0.0.1'))
     72 
     73 endtime = time.time() + TIMEOUT
     74 sock.sendto(payload, (MCAST_GRP, MCAST_PORT))
     75 
     76 while time.time() < endtime:
     77   try:
     78     data, addr = sock.recvfrom(1024)
     79   except socket.error as e:
     80     print('recv exception: ', e)
     81   else:
     82     reply = Ether(data)
     83     if is_dhcp_offer(reply):
     84         opts = dict([o for o in reply[DHCP].options if type(o) is tuple])
     85         print(json.dumps(opts, cls=JSONEncoderWithBytes))
     86         exit(0)
     87 exit(1)