liminix/tests/pppoe/test-dhcp-service.py

89 lines
2.3 KiB
Python
Raw Normal View History

# forge packets for testing liminix and send them via the qemu udp
# multicast socket interface
MCAST_GRP = '230.0.0.1'
MCAST_PORT = 1235
MULTICAST_TTL = 2
TIMEOUT = 10 # seconds
from warnings import filterwarnings
filterwarnings("ignore")
import random
import binascii
import socket
import time
import json
from builtins import bytes, bytearray
from scapy.all import Ether, IP, UDP, BOOTP, DHCP, sendp, send, raw
class JSONEncoderWithBytes(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (bytes, bytearray)):
return obj.decode('utf-8')
return json.JSONEncoder.default(self, obj)
def dhcp_option(pkt, label):
if pkt.haslayer(DHCP):
for i in pkt[DHCP].options:
l, v = i
if l == label:
return v
return None
def is_dhcp_offer(pkt):
val = dhcp_option(pkt, 'message-type')
return (val == 2)
def mac_to_bytes(mac_addr: str) -> bytes:
""" Converts a MAC address string to bytes.
"""
return int(mac_addr.replace(":", ""), 16).to_bytes(6, "big")
client_mac = "01:02:03:04:05:06"
discover = (
Ether(dst="ff:ff:ff:ff:ff:ff") /
IP(src="0.0.0.0", dst="255.255.255.255") /
UDP(sport=68, dport=67) /
BOOTP(
chaddr=mac_to_bytes(client_mac),
xid=random.randint(1, 2**32-1),
) /
DHCP(options=[("message-type", "discover"), "end"])
)
payload = raw(discover)
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.settimeout(TIMEOUT)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, MULTICAST_TTL)
sock.bind((MCAST_GRP, MCAST_PORT))
host = socket.gethostbyname(socket.gethostname())
sock.setsockopt(socket.SOL_IP, socket.IP_MULTICAST_IF, socket.inet_aton(host))
sock.setsockopt(socket.SOL_IP, socket.IP_ADD_MEMBERSHIP,
socket.inet_aton(MCAST_GRP) + socket.inet_aton(host))
endtime = time.time() + TIMEOUT
sock.sendto(payload, (MCAST_GRP, MCAST_PORT))
while time.time() < endtime:
try:
data, addr = sock.recvfrom(1024)
except socket.error:
print('Exception')
else:
reply = Ether(data)
if is_dhcp_offer(reply):
opts = dict([o for o in reply[DHCP].options if type(o) is tuple])
print(json.dumps(opts, cls=JSONEncoderWithBytes))
exit(0)
exit(1)