Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/scapy/layers/dhcp.py: 47%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# SPDX-License-Identifier: GPL-2.0-only
2# This file is part of Scapy
3# See https://scapy.net/ for more information
4# Copyright (C) Philippe Biondi <phil@secdev.org>
6"""
7DHCP (Dynamic Host Configuration Protocol) and BOOTP
9Implements:
10- rfc951 - BOOTSTRAP PROTOCOL (BOOTP)
11- rfc1542 - Clarifications and Extensions for the Bootstrap Protocol
12- rfc1533 - DHCP Options and BOOTP Vendor Extensions
13"""
15try:
16 from collections.abc import Iterable
17except ImportError:
18 # For backwards compatibility. This was removed in Python 3.8
19 from collections import Iterable
20import random
21import struct
23import socket
24import re
26from scapy.ansmachine import AnsweringMachine
27from scapy.base_classes import Net
28from scapy.compat import chb, orb, bytes_encode
29from scapy.fields import (
30 ByteEnumField,
31 ByteField,
32 Field,
33 FieldListField,
34 FlagsField,
35 IntField,
36 IPField,
37 ShortField,
38 StrEnumField,
39 StrField,
40 StrFixedLenField,
41 XIntField,
42)
43from scapy.layers.inet import UDP, IP
44from scapy.layers.l2 import Ether, HARDWARE_TYPES
45from scapy.packet import bind_layers, bind_bottom_up, Packet
46from scapy.utils import atol, itom, ltoa, sane, str2mac
47from scapy.volatile import (
48 RandBin,
49 RandByte,
50 RandField,
51 RandIP,
52 RandInt,
53 RandNum,
54 RandNumExpo,
55)
57from scapy.arch import get_if_raw_hwaddr
58from scapy.sendrecv import srp1
59from scapy.error import warning
60from scapy.config import conf
62dhcpmagic = b"c\x82Sc"
65class _BOOTP_chaddr(StrFixedLenField):
66 def i2repr(self, pkt, v):
67 if pkt.htype == 1: # Ethernet
68 if v[6:] == b"\x00" * 10: # Default padding
69 return "%s (+ 10 nul pad)" % str2mac(v[:6])
70 else:
71 return "%s (pad: %s)" % (str2mac(v[:6]), v[6:])
72 return super(_BOOTP_chaddr, self).i2repr(pkt, v)
75class BOOTP(Packet):
76 name = "BOOTP"
77 fields_desc = [
78 ByteEnumField("op", 1, {1: "BOOTREQUEST", 2: "BOOTREPLY"}),
79 ByteEnumField("htype", 1, HARDWARE_TYPES),
80 ByteField("hlen", 6),
81 ByteField("hops", 0),
82 XIntField("xid", 0),
83 ShortField("secs", 0),
84 FlagsField("flags", 0, 16, "???????????????B"),
85 IPField("ciaddr", "0.0.0.0"),
86 IPField("yiaddr", "0.0.0.0"),
87 IPField("siaddr", "0.0.0.0"),
88 IPField("giaddr", "0.0.0.0"),
89 _BOOTP_chaddr("chaddr", b"", length=16),
90 StrFixedLenField("sname", b"", length=64),
91 StrFixedLenField("file", b"", length=128),
92 StrEnumField("options", b"", {dhcpmagic: "DHCP magic"})]
94 def guess_payload_class(self, payload):
95 if self.options[:len(dhcpmagic)] == dhcpmagic:
96 return DHCP
97 else:
98 return Packet.guess_payload_class(self, payload)
100 def extract_padding(self, s):
101 if self.options[:len(dhcpmagic)] == dhcpmagic:
102 # set BOOTP options to DHCP magic cookie and make rest a payload of DHCP options # noqa: E501
103 payload = self.options[len(dhcpmagic):]
104 self.options = self.options[:len(dhcpmagic)]
105 return payload, None
106 else:
107 return b"", None
109 def hashret(self):
110 return struct.pack("!I", self.xid)
112 def answers(self, other):
113 if not isinstance(other, BOOTP):
114 return 0
115 return self.xid == other.xid
118class _DHCPParamReqFieldListField(FieldListField):
119 def randval(self):
120 class _RandReqFieldList(RandField):
121 def _fix(self):
122 return [RandByte()] * int(RandByte())
123 return _RandReqFieldList()
126class RandClasslessStaticRoutesField(RandField):
127 """
128 A RandValue for classless static routes
129 """
131 def _fix(self):
132 return "%s/%d:%s" % (RandIP(), RandNum(0, 32), RandIP())
135class ClasslessFieldListField(FieldListField):
136 def randval(self):
137 class _RandClasslessField(RandField):
138 def _fix(self):
139 return [RandClasslessStaticRoutesField()] * int(RandNum(1, 28))
140 return _RandClasslessField()
143class ClasslessStaticRoutesField(Field):
144 """
145 RFC 3442 defines classless static routes as up to 9 bytes per entry:
147 # Code Len Destination 1 Router 1
148 +-----+---+----+-----+----+----+----+----+----+
149 | 121 | n | d1 | ... | dN | r1 | r2 | r3 | r4 |
150 +-----+---+----+-----+----+----+----+----+----+
152 Destination first byte contains one octet describing the width followed
153 by all the significant octets of the subnet.
154 """
156 def m2i(self, pkt, x):
157 # type: (Packet, bytes) -> str
158 # b'\x20\x01\x02\x03\x04\t\x08\x07\x06' -> (1.2.3.4/32:9.8.7.6)
159 prefix = orb(x[0])
161 octets = (prefix + 7) // 8
162 # Create the destination IP by using the number of octets
163 # and padding up to 4 bytes to ensure a valid IP.
164 dest = x[1:1 + octets]
165 dest = socket.inet_ntoa(dest.ljust(4, b'\x00'))
167 router = x[1 + octets:5 + octets]
168 router = socket.inet_ntoa(router)
170 return dest + "/" + str(prefix) + ":" + router
172 def i2m(self, pkt, x):
173 # type: (Packet, str) -> bytes
174 # (1.2.3.4/32:9.8.7.6) -> b'\x20\x01\x02\x03\x04\t\x08\x07\x06'
175 if not x:
176 return b''
178 spx = re.split('/|:', str(x))
179 prefix = int(spx[1])
180 # if prefix is invalid value ( 0 > prefix > 32 ) then break
181 if prefix > 32 or prefix < 0:
182 warning("Invalid prefix value: %d (0x%x)", prefix, prefix)
183 return b''
184 octets = (prefix + 7) // 8
185 dest = socket.inet_aton(spx[0])[:octets]
186 router = socket.inet_aton(spx[2])
187 return struct.pack('b', prefix) + dest + router
189 def getfield(self, pkt, s):
190 prefix = orb(s[0])
191 route_len = 5 + (prefix + 7) // 8
192 return s[route_len:], self.m2i(pkt, s[:route_len])
194 def addfield(self, pkt, s, val):
195 return s + self.i2m(pkt, val)
197 def randval(self):
198 return RandClasslessStaticRoutesField()
201# DHCP_UNKNOWN, DHCP_IP, DHCP_IPLIST, DHCP_TYPE \
202# = range(4)
203#
205# DHCP Options and BOOTP Vendor Extensions
208DHCPTypes = {
209 1: "discover",
210 2: "offer",
211 3: "request",
212 4: "decline",
213 5: "ack",
214 6: "nak",
215 7: "release",
216 8: "inform",
217 9: "force_renew",
218 10: "lease_query",
219 11: "lease_unassigned",
220 12: "lease_unknown",
221 13: "lease_active",
222}
224DHCPOptions = {
225 0: "pad",
226 1: IPField("subnet_mask", "0.0.0.0"),
227 2: IntField("time_zone", 500),
228 3: IPField("router", "0.0.0.0"),
229 4: IPField("time_server", "0.0.0.0"),
230 5: IPField("IEN_name_server", "0.0.0.0"),
231 6: IPField("name_server", "0.0.0.0"),
232 7: IPField("log_server", "0.0.0.0"),
233 8: IPField("cookie_server", "0.0.0.0"),
234 9: IPField("lpr_server", "0.0.0.0"),
235 10: IPField("impress-servers", "0.0.0.0"),
236 11: IPField("resource-location-servers", "0.0.0.0"),
237 12: "hostname",
238 13: ShortField("boot-size", 1000),
239 14: "dump_path",
240 15: "domain",
241 16: IPField("swap-server", "0.0.0.0"),
242 17: "root_disk_path",
243 18: "extensions-path",
244 19: ByteField("ip-forwarding", 0),
245 20: ByteField("non-local-source-routing", 0),
246 21: IPField("policy-filter", "0.0.0.0"),
247 22: ShortField("max_dgram_reass_size", 300),
248 23: ByteField("default_ttl", 50),
249 24: IntField("pmtu_timeout", 1000),
250 25: ShortField("path-mtu-plateau-table", 1000),
251 26: ShortField("interface-mtu", 50),
252 27: ByteField("all-subnets-local", 0),
253 28: IPField("broadcast_address", "0.0.0.0"),
254 29: ByteField("perform-mask-discovery", 0),
255 30: ByteField("mask-supplier", 0),
256 31: ByteField("router-discovery", 0),
257 32: IPField("router-solicitation-address", "0.0.0.0"),
258 33: IPField("static-routes", "0.0.0.0"),
259 34: ByteField("trailer-encapsulation", 0),
260 35: IntField("arp_cache_timeout", 1000),
261 36: ByteField("ieee802-3-encapsulation", 0),
262 37: ByteField("tcp_ttl", 100),
263 38: IntField("tcp_keepalive_interval", 1000),
264 39: ByteField("tcp_keepalive_garbage", 0),
265 40: StrField("NIS_domain", "www.example.com"),
266 41: IPField("NIS_server", "0.0.0.0"),
267 42: IPField("NTP_server", "0.0.0.0"),
268 43: "vendor_specific",
269 44: IPField("NetBIOS_server", "0.0.0.0"),
270 45: IPField("NetBIOS_dist_server", "0.0.0.0"),
271 46: ByteField("NetBIOS_node_type", 100),
272 47: "netbios-scope",
273 48: IPField("font-servers", "0.0.0.0"),
274 49: IPField("x-display-manager", "0.0.0.0"),
275 50: IPField("requested_addr", "0.0.0.0"),
276 51: IntField("lease_time", 43200),
277 52: ByteField("dhcp-option-overload", 100),
278 53: ByteEnumField("message-type", 1, DHCPTypes),
279 54: IPField("server_id", "0.0.0.0"),
280 55: _DHCPParamReqFieldListField(
281 "param_req_list", [],
282 ByteField("opcode", 0)),
283 56: "error_message",
284 57: ShortField("max_dhcp_size", 1500),
285 58: IntField("renewal_time", 21600),
286 59: IntField("rebinding_time", 37800),
287 60: StrField("vendor_class_id", "id"),
288 61: StrField("client_id", ""),
289 62: "nwip-domain-name",
290 64: "NISplus_domain",
291 65: IPField("NISplus_server", "0.0.0.0"),
292 66: "tftp_server_name",
293 67: StrField("boot-file-name", ""),
294 68: IPField("mobile-ip-home-agent", "0.0.0.0"),
295 69: IPField("SMTP_server", "0.0.0.0"),
296 70: IPField("POP3_server", "0.0.0.0"),
297 71: IPField("NNTP_server", "0.0.0.0"),
298 72: IPField("WWW_server", "0.0.0.0"),
299 73: IPField("Finger_server", "0.0.0.0"),
300 74: IPField("IRC_server", "0.0.0.0"),
301 75: IPField("StreetTalk_server", "0.0.0.0"),
302 76: IPField("StreetTalk_Dir_Assistance", "0.0.0.0"),
303 77: "user_class",
304 78: "slp_service_agent",
305 79: "slp_service_scope",
306 80: "rapid_commit",
307 81: "client_FQDN",
308 82: "relay_agent_information",
309 85: IPField("nds-server", "0.0.0.0"),
310 86: StrField("nds-tree-name", ""),
311 87: StrField("nds-context", ""),
312 88: "bcms-controller-namesi",
313 89: IPField("bcms-controller-address", "0.0.0.0"),
314 91: IntField("client-last-transaction-time", 1000),
315 92: IPField("associated-ip", "0.0.0.0"),
316 93: "pxe_client_architecture",
317 94: "pxe_client_network_interface",
318 97: "pxe_client_machine_identifier",
319 98: StrField("uap-servers", ""),
320 100: StrField("pcode", ""),
321 101: StrField("tcode", ""),
322 108: IntField("ipv6-only-preferred", 0),
323 112: IPField("netinfo-server-address", "0.0.0.0"),
324 113: StrField("netinfo-server-tag", ""),
325 114: StrField("captive-portal", ""),
326 116: ByteField("auto-config", 0),
327 117: ShortField("name-service-search", 0,),
328 118: IPField("subnet-selection", "0.0.0.0"),
329 121: ClasslessFieldListField(
330 "classless_static_routes",
331 [],
332 ClasslessStaticRoutesField("route", 0)),
333 124: "vendor_class",
334 125: "vendor_specific_information",
335 128: IPField("tftp_server_ip_address", "0.0.0.0"),
336 136: IPField("pana-agent", "0.0.0.0"),
337 137: "v4-lost",
338 138: IPField("capwap-ac-v4", "0.0.0.0"),
339 141: "sip_ua_service_domains",
340 146: "rdnss-selection",
341 150: IPField("tftp_server_address", "0.0.0.0"),
342 159: "v4-portparams",
343 160: StrField("v4-captive-portal", ""),
344 161: StrField("mud-url", ""),
345 208: "pxelinux_magic",
346 209: "pxelinux_configuration_file",
347 210: "pxelinux_path_prefix",
348 211: "pxelinux_reboot_time",
349 212: "option-6rd",
350 213: "v4-access-domain",
351 255: "end"
352}
354DHCPRevOptions = {}
356for k, v in DHCPOptions.items():
357 if isinstance(v, str):
358 n = v
359 v = None
360 else:
361 n = v.name
362 DHCPRevOptions[n] = (k, v)
363del n
364del v
365del k
368class RandDHCPOptions(RandField):
369 def __init__(self, size=None, rndstr=None):
370 if size is None:
371 size = RandNumExpo(0.05)
372 self.size = size
373 if rndstr is None:
374 rndstr = RandBin(RandNum(0, 255))
375 self.rndstr = rndstr
376 self._opts = list(DHCPOptions.values())
377 self._opts.remove("pad")
378 self._opts.remove("end")
380 def _fix(self):
381 op = []
382 for k in range(self.size):
383 o = random.choice(self._opts)
384 if isinstance(o, str):
385 op.append((o, self.rndstr * 1))
386 else:
387 r = o.randval()._fix()
388 if isinstance(r, bytes):
389 r = r[:255]
390 op.append((o.name, r))
391 return op
394class DHCPOptionsField(StrField):
395 """
396 A field that builds and dissects DHCP options.
397 The internal value is a list of tuples with the format
398 [("option_name", <option_value>), ...]
399 Where expected names and values can be found using `DHCPOptions`
400 """
401 islist = 1
403 def i2repr(self, pkt, x):
404 s = []
405 for v in x:
406 if isinstance(v, tuple) and len(v) >= 2:
407 if v[0] in DHCPRevOptions and isinstance(DHCPRevOptions[v[0]][1], Field): # noqa: E501
408 f = DHCPRevOptions[v[0]][1]
409 vv = ",".join(f.i2repr(pkt, val) for val in v[1:])
410 else:
411 vv = ",".join(repr(val) for val in v[1:])
412 s.append("%s=%s" % (v[0], vv))
413 else:
414 s.append(sane(v))
415 return "[%s]" % (" ".join(s))
417 def getfield(self, pkt, s):
418 return b"", self.m2i(pkt, s)
420 def m2i(self, pkt, x):
421 opt = []
422 while x:
423 o = orb(x[0])
424 if o == 255:
425 opt.append("end")
426 x = x[1:]
427 continue
428 if o == 0:
429 opt.append("pad")
430 x = x[1:]
431 continue
432 if len(x) < 2 or len(x) < orb(x[1]) + 2:
433 opt.append(x)
434 break
435 elif o in DHCPOptions:
436 f = DHCPOptions[o]
438 if isinstance(f, str):
439 olen = orb(x[1])
440 opt.append((f, x[2:olen + 2]))
441 x = x[olen + 2:]
442 else:
443 olen = orb(x[1])
444 lval = [f.name]
446 if olen == 0:
447 try:
448 _, val = f.getfield(pkt, b'')
449 except Exception:
450 opt.append(x)
451 break
452 else:
453 lval.append(val)
455 try:
456 left = x[2:olen + 2]
457 while left:
458 left, val = f.getfield(pkt, left)
459 lval.append(val)
460 except Exception:
461 opt.append(x)
462 break
463 else:
464 otuple = tuple(lval)
465 opt.append(otuple)
466 x = x[olen + 2:]
467 else:
468 olen = orb(x[1])
469 opt.append((o, x[2:olen + 2]))
470 x = x[olen + 2:]
471 return opt
473 def i2m(self, pkt, x):
474 if isinstance(x, str):
475 return x
476 s = b""
477 for o in x:
478 if isinstance(o, tuple) and len(o) >= 2:
479 name = o[0]
480 lval = o[1:]
482 if isinstance(name, int):
483 onum, oval = name, b"".join(lval)
484 elif name in DHCPRevOptions:
485 onum, f = DHCPRevOptions[name]
486 if f is not None:
487 lval = (f.addfield(pkt, b"", f.any2i(pkt, val)) for val in lval) # noqa: E501
488 else:
489 lval = (bytes_encode(x) for x in lval)
490 oval = b"".join(lval)
491 else:
492 warning("Unknown field option %s", name)
493 continue
495 s += struct.pack("!BB", onum, len(oval))
496 s += oval
498 elif (isinstance(o, str) and o in DHCPRevOptions and
499 DHCPRevOptions[o][1] is None):
500 s += chb(DHCPRevOptions[o][0])
501 elif isinstance(o, int):
502 s += chb(o) + b"\0"
503 elif isinstance(o, (str, bytes)):
504 s += bytes_encode(o)
505 else:
506 warning("Malformed option %s", o)
507 return s
509 def randval(self):
510 return RandDHCPOptions()
513class DHCP(Packet):
514 name = "DHCP options"
515 fields_desc = [DHCPOptionsField("options", b"")]
517 def mysummary(self):
518 for id in self.options:
519 if isinstance(id, tuple) and id[0] == "message-type":
520 return "DHCP %s" % DHCPTypes.get(id[1], "").capitalize()
521 return super(DHCP, self).mysummary()
524bind_layers(UDP, BOOTP, dport=67, sport=68)
525bind_layers(UDP, BOOTP, dport=68, sport=67)
526bind_bottom_up(UDP, BOOTP, dport=67, sport=67)
527bind_layers(BOOTP, DHCP, options=b'c\x82Sc')
530@conf.commands.register
531def dhcp_request(hw=None,
532 req_type='discover',
533 server_id=None,
534 requested_addr=None,
535 hostname=None,
536 iface=None,
537 **kargs):
538 """
539 Send a DHCP discover request and return the answer.
541 Usage::
543 >>> dhcp_request() # send DHCP discover
544 >>> dhcp_request(req_type='request',
545 ... requested_addr='10.53.4.34') # send DHCP request
546 """
547 if conf.checkIPaddr:
548 warning(
549 "conf.checkIPaddr is enabled, may not be able to match the answer"
550 )
551 if hw is None:
552 if iface is None:
553 iface = conf.iface
554 _, hw = get_if_raw_hwaddr(iface)
555 dhcp_options = [
556 ('message-type', req_type),
557 ('client_id', b'\x01' + hw),
558 ]
559 if requested_addr is not None:
560 dhcp_options.append(('requested_addr', requested_addr))
561 elif req_type == 'request':
562 warning("DHCP Request without requested_addr will likely be ignored")
563 if server_id is not None:
564 dhcp_options.append(('server_id', server_id))
565 if hostname is not None:
566 dhcp_options.extend([
567 ('hostname', hostname),
568 ('client_FQDN', b'\x00\x00\x00' + bytes_encode(hostname)),
569 ])
570 dhcp_options.extend([
571 ('vendor_class_id', b'MSFT 5.0'),
572 ('param_req_list', [
573 1, 3, 6, 15, 31, 33, 43, 44, 46, 47, 119, 121, 249, 252
574 ]),
575 'end'
576 ])
577 return srp1(
578 Ether(dst="ff:ff:ff:ff:ff:ff", src=hw) /
579 IP(src="0.0.0.0", dst="255.255.255.255") /
580 UDP(sport=68, dport=67) /
581 BOOTP(chaddr=hw, xid=RandInt(), flags="B") /
582 DHCP(options=dhcp_options),
583 iface=iface, **kargs
584 )
587class BOOTP_am(AnsweringMachine):
588 function_name = "bootpd"
589 filter = "udp and port 68 and port 67"
591 def parse_options(self,
592 pool=Net("192.168.1.128/25"),
593 network="192.168.1.0/24",
594 gw="192.168.1.1",
595 nameserver=None,
596 domain=None,
597 renewal_time=60,
598 lease_time=1800,
599 **kwargs):
600 """
601 :param pool: the range of addresses to distribute. Can be a Net,
602 a list of IPs or a string (always gives the same IP).
603 :param network: the subnet range
604 :param gw: the gateway IP (can be None)
605 :param nameserver: the DNS server IP (by default, same than gw)
606 :param domain: the domain to advertise (can be None)
608 Other DHCP parameters can be passed as kwargs. See DHCPOptions in dhcp.py.
609 For instance::
611 dhcpd(pool=Net("10.0.10.0/24"), network="10.0.0.0/8", gw="10.0.10.1",
612 classless_static_routes=["1.2.3.4/32:9.8.7.6"])
613 """
614 self.domain = domain
615 netw, msk = (network.split("/") + ["32"])[:2]
616 msk = itom(int(msk))
617 self.netmask = ltoa(msk)
618 self.network = ltoa(atol(netw) & msk)
619 self.broadcast = ltoa(atol(self.network) | (0xffffffff & ~msk))
620 self.gw = gw
621 self.nameserver = nameserver or gw
622 if isinstance(pool, str):
623 pool = Net(pool)
624 if isinstance(pool, Iterable):
625 pool = [k for k in pool if k not in [gw, self.network, self.broadcast]]
626 pool.reverse()
627 if len(pool) == 1:
628 pool, = pool
629 self.pool = pool
630 self.lease_time = lease_time
631 self.renewal_time = renewal_time
632 self.leases = {}
633 self.kwargs = kwargs
635 def is_request(self, req):
636 if not req.haslayer(BOOTP):
637 return 0
638 reqb = req.getlayer(BOOTP)
639 if reqb.op != 1:
640 return 0
641 return 1
643 def print_reply(self, _, reply):
644 print("Reply %s to %s" % (reply.getlayer(IP).dst, reply.dst))
646 def make_reply(self, req):
647 mac = req[Ether].src
648 if isinstance(self.pool, list):
649 if mac not in self.leases:
650 self.leases[mac] = self.pool.pop()
651 ip = self.leases[mac]
652 else:
653 ip = self.pool
655 repb = req.getlayer(BOOTP).copy()
656 repb.op = "BOOTREPLY"
657 repb.yiaddr = ip
658 repb.siaddr = self.gw
659 repb.ciaddr = self.gw
660 repb.giaddr = self.gw
661 del repb.payload
662 rep = Ether(dst=mac) / IP(dst=ip) / UDP(sport=req.dport, dport=req.sport) / repb # noqa: E501
663 return rep
666class DHCP_am(BOOTP_am):
667 function_name = "dhcpd"
669 def make_reply(self, req):
670 resp = BOOTP_am.make_reply(self, req)
671 if DHCP in req:
672 dhcp_options = [
673 (op[0], {1: 2, 3: 5}.get(op[1], op[1]))
674 for op in req[DHCP].options
675 if isinstance(op, tuple) and op[0] == "message-type"
676 ]
677 dhcp_options += [
678 x for x in [
679 ("server_id", self.gw),
680 ("domain", self.domain),
681 ("router", self.gw),
682 ("name_server", self.nameserver),
683 ("broadcast_address", self.broadcast),
684 ("subnet_mask", self.netmask),
685 ("renewal_time", self.renewal_time),
686 ("lease_time", self.lease_time),
687 ]
688 if x[1] is not None
689 ]
690 if self.kwargs:
691 dhcp_options += self.kwargs.items()
692 dhcp_options.append("end")
693 resp /= DHCP(options=dhcp_options)
694 return resp