| ## This file is part of Scapy |
| ## See http://www.secdev.org/projects/scapy for more informations |
| ## Copyright (C) Philippe Biondi <phil@secdev.org> |
| ## This program is published under a GPLv2 license |
| |
| """ |
| DHCP (Dynamic Host Configuration Protocol) and BOOTP |
| """ |
| |
| from __future__ import absolute_import |
| from __future__ import print_function |
| from collections import Iterable |
| import struct |
| |
| from scapy.packet import * |
| from scapy.fields import * |
| from scapy.ansmachine import * |
| from scapy.data import * |
| from scapy.compat import * |
| from scapy.layers.inet import UDP,IP |
| from scapy.layers.l2 import Ether |
| from scapy.base_classes import Net |
| from scapy.volatile import RandField |
| |
| from scapy.arch import get_if_raw_hwaddr |
| from scapy.sendrecv import * |
| from scapy.error import warning |
| import scapy.modules.six as six |
| from scapy.modules.six.moves import range |
| |
| dhcpmagic=b"c\x82Sc" |
| |
| |
| class BOOTP(Packet): |
| name = "BOOTP" |
| fields_desc = [ ByteEnumField("op",1, {1:"BOOTREQUEST", 2:"BOOTREPLY"}), |
| ByteField("htype",1), |
| ByteField("hlen",6), |
| ByteField("hops",0), |
| IntField("xid",0), |
| ShortField("secs",0), |
| FlagsField("flags", 0, 16, "???????????????B"), |
| IPField("ciaddr","0.0.0.0"), |
| IPField("yiaddr","0.0.0.0"), |
| IPField("siaddr","0.0.0.0"), |
| IPField("giaddr","0.0.0.0"), |
| Field("chaddr",b"", "16s"), |
| Field("sname",b"","64s"), |
| Field("file",b"","128s"), |
| StrField("options",b"") ] |
| def guess_payload_class(self, payload): |
| if self.options[:len(dhcpmagic)] == dhcpmagic: |
| return DHCP |
| else: |
| return Packet.guess_payload_class(self, payload) |
| def extract_padding(self,s): |
| if self.options[:len(dhcpmagic)] == dhcpmagic: |
| # set BOOTP options to DHCP magic cookie and make rest a payload of DHCP options |
| payload = self.options[len(dhcpmagic):] |
| self.options = self.options[:len(dhcpmagic)] |
| return payload, None |
| else: |
| return b"", None |
| def hashret(self): |
| return struct.pack("L", self.xid) |
| def answers(self, other): |
| if not isinstance(other, BOOTP): |
| return 0 |
| return self.xid == other.xid |
| |
| |
| class _DHCPParamReqFieldListField(FieldListField): |
| def getfield(self, pkt, s): |
| ret = [] |
| while s: |
| s, val = FieldListField.getfield(self, pkt, s) |
| ret.append(val) |
| return b"", [x[0] for x in ret] |
| |
| #DHCP_UNKNOWN, DHCP_IP, DHCP_IPLIST, DHCP_TYPE \ |
| #= range(4) |
| # |
| |
| DHCPTypes = { |
| 1: "discover", |
| 2: "offer", |
| 3: "request", |
| 4: "decline", |
| 5: "ack", |
| 6: "nak", |
| 7: "release", |
| 8: "inform", |
| 9: "force_renew", |
| 10:"lease_query", |
| 11:"lease_unassigned", |
| 12:"lease_unknown", |
| 13:"lease_active", |
| } |
| |
| DHCPOptions = { |
| 0: "pad", |
| 1: IPField("subnet_mask", "0.0.0.0"), |
| 2: "time_zone", |
| 3: IPField("router","0.0.0.0"), |
| 4: IPField("time_server","0.0.0.0"), |
| 5: IPField("IEN_name_server","0.0.0.0"), |
| 6: IPField("name_server","0.0.0.0"), |
| 7: IPField("log_server","0.0.0.0"), |
| 8: IPField("cookie_server","0.0.0.0"), |
| 9: IPField("lpr_server","0.0.0.0"), |
| 12: "hostname", |
| 14: "dump_path", |
| 15: "domain", |
| 17: "root_disk_path", |
| 22: "max_dgram_reass_size", |
| 23: "default_ttl", |
| 24: "pmtu_timeout", |
| 28: IPField("broadcast_address","0.0.0.0"), |
| 35: "arp_cache_timeout", |
| 36: "ether_or_dot3", |
| 37: "tcp_ttl", |
| 38: "tcp_keepalive_interval", |
| 39: "tcp_keepalive_garbage", |
| 40: "NIS_domain", |
| 41: IPField("NIS_server","0.0.0.0"), |
| 42: IPField("NTP_server","0.0.0.0"), |
| 43: "vendor_specific", |
| 44: IPField("NetBIOS_server","0.0.0.0"), |
| 45: IPField("NetBIOS_dist_server","0.0.0.0"), |
| 50: IPField("requested_addr","0.0.0.0"), |
| 51: IntField("lease_time", 43200), |
| 53: ByteEnumField("message-type", 1, DHCPTypes), |
| 54: IPField("server_id","0.0.0.0"), |
| 55: _DHCPParamReqFieldListField("param_req_list", [], ByteField("opcode", 0), length_from=lambda x: 1), |
| 56: "error_message", |
| 57: ShortField("max_dhcp_size", 1500), |
| 58: IntField("renewal_time", 21600), |
| 59: IntField("rebinding_time", 37800), |
| 60: "vendor_class_id", |
| 61: "client_id", |
| |
| 64: "NISplus_domain", |
| 65: IPField("NISplus_server","0.0.0.0"), |
| 69: IPField("SMTP_server","0.0.0.0"), |
| 70: IPField("POP3_server","0.0.0.0"), |
| 71: IPField("NNTP_server","0.0.0.0"), |
| 72: IPField("WWW_server","0.0.0.0"), |
| 73: IPField("Finger_server","0.0.0.0"), |
| 74: IPField("IRC_server","0.0.0.0"), |
| 75: IPField("StreetTalk_server","0.0.0.0"), |
| 76: "StreetTalk_Dir_Assistance", |
| 82: "relay_agent_Information", |
| 255: "end" |
| } |
| |
| DHCPRevOptions = {} |
| |
| for k,v in six.iteritems(DHCPOptions): |
| if isinstance(v, str): |
| n = v |
| v = None |
| else: |
| n = v.name |
| DHCPRevOptions[n] = (k,v) |
| del(n) |
| del(v) |
| del(k) |
| |
| |
| |
| |
| class RandDHCPOptions(RandField): |
| def __init__(self, size=None, rndstr=None): |
| if size is None: |
| size = RandNumExpo(0.05) |
| self.size = size |
| if rndstr is None: |
| rndstr = RandBin(RandNum(0,255)) |
| self.rndstr=rndstr |
| self._opts = list(DHCPOptions.values()) |
| self._opts.remove("pad") |
| self._opts.remove("end") |
| def _fix(self): |
| op = [] |
| for k in range(self.size): |
| o = random.choice(self._opts) |
| if isinstance(o, str): |
| op.append((o,self.rndstr*1)) |
| else: |
| op.append((o.name, o.randval()._fix())) |
| return op |
| |
| |
| class DHCPOptionsField(StrField): |
| islist=1 |
| def i2repr(self,pkt,x): |
| s = [] |
| for v in x: |
| if isinstance(v, tuple) and len(v) >= 2: |
| if v[0] in DHCPRevOptions and isinstance(DHCPRevOptions[v[0]][1],Field): |
| f = DHCPRevOptions[v[0]][1] |
| vv = ",".join(f.i2repr(pkt,val) for val in v[1:]) |
| else: |
| vv = ",".join(repr(val) for val in v[1:]) |
| r = "%s=%s" % (v[0],vv) |
| s.append(r) |
| else: |
| s.append(sane(v)) |
| return "[%s]" % (" ".join(s)) |
| |
| def getfield(self, pkt, s): |
| return b"", self.m2i(pkt, s) |
| def m2i(self, pkt, x): |
| opt = [] |
| while x: |
| o = orb(x[0]) |
| if o == 255: |
| opt.append("end") |
| x = x[1:] |
| continue |
| if o == 0: |
| opt.append("pad") |
| x = x[1:] |
| continue |
| if len(x) < 2 or len(x) < orb(x[1])+2: |
| opt.append(x) |
| break |
| elif o in DHCPOptions: |
| f = DHCPOptions[o] |
| |
| if isinstance(f, str): |
| olen = orb(x[1]) |
| opt.append( (f,x[2:olen+2]) ) |
| x = x[olen+2:] |
| else: |
| olen = orb(x[1]) |
| lval = [f.name] |
| try: |
| left = x[2:olen+2] |
| while left: |
| left, val = f.getfield(pkt,left) |
| lval.append(val) |
| except: |
| opt.append(x) |
| break |
| else: |
| otuple = tuple(lval) |
| opt.append(otuple) |
| x = x[olen+2:] |
| else: |
| olen = orb(x[1]) |
| opt.append((o, x[2:olen+2])) |
| x = x[olen+2:] |
| return opt |
| def i2m(self, pkt, x): |
| if isinstance(x, str): |
| return x |
| s = b"" |
| for o in x: |
| if isinstance(o, tuple) and len(o) >= 2: |
| name = o[0] |
| lval = o[1:] |
| |
| if isinstance(name, int): |
| onum, oval = name, b"".join(lval) |
| elif name in DHCPRevOptions: |
| onum, f = DHCPRevOptions[name] |
| if f is not None: |
| lval = [f.addfield(pkt,b"",f.any2i(pkt,val)) for val in lval] |
| oval = b"".join(lval) |
| else: |
| warning("Unknown field option %s", name) |
| continue |
| |
| s += chb(onum) |
| s += chb(len(oval)) |
| s += oval |
| |
| elif (isinstance(o, str) and o in DHCPRevOptions and |
| DHCPRevOptions[o][1] == None): |
| s += chb(DHCPRevOptions[o][0]) |
| elif isinstance(o, int): |
| s += chb(o)+b"\0" |
| elif isinstance(o, (str, bytes)): |
| s += raw(o) |
| else: |
| warning("Malformed option %s", o) |
| return s |
| |
| |
| class DHCP(Packet): |
| name = "DHCP options" |
| fields_desc = [ DHCPOptionsField("options",b"") ] |
| |
| |
| bind_layers( UDP, BOOTP, dport=67, sport=68) |
| bind_layers( UDP, BOOTP, dport=68, sport=67) |
| bind_bottom_up( UDP, BOOTP, dport=67, sport=67) |
| bind_layers( BOOTP, DHCP, options=b'c\x82Sc') |
| |
| @conf.commands.register |
| def dhcp_request(iface=None,**kargs): |
| if conf.checkIPaddr != 0: |
| warning("conf.checkIPaddr is not 0, I may not be able to match the answer") |
| if iface is None: |
| iface = conf.iface |
| fam,hw = get_if_raw_hwaddr(iface) |
| return srp1(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=hw)/DHCP(options=[("message-type","discover"),"end"]),iface=iface,**kargs) |
| |
| |
| class BOOTP_am(AnsweringMachine): |
| function_name = "bootpd" |
| filter = "udp and port 68 and port 67" |
| send_function = staticmethod(sendp) |
| def parse_options(self, pool=Net("192.168.1.128/25"), network="192.168.1.0/24",gw="192.168.1.1", |
| domain="localnet", renewal_time=60, lease_time=1800): |
| self.domain = domain |
| netw,msk = (network.split("/")+["32"])[:2] |
| msk = itom(int(msk)) |
| self.netmask = ltoa(msk) |
| self.network = ltoa(atol(netw)&msk) |
| self.broadcast = ltoa( atol(self.network) | (0xffffffff&~msk) ) |
| self.gw = gw |
| if isinstance(pool, six.string_types): |
| pool = Net(pool) |
| if isinstance(pool, Iterable): |
| pool = [k for k in pool if k not in [gw, self.network, self.broadcast]] |
| pool.reverse() |
| if len(pool) == 1: |
| pool, = pool |
| self.pool = pool |
| self.lease_time = lease_time |
| self.renewal_time = renewal_time |
| self.leases = {} |
| |
| def is_request(self, req): |
| if not req.haslayer(BOOTP): |
| return 0 |
| reqb = req.getlayer(BOOTP) |
| if reqb.op != 1: |
| return 0 |
| return 1 |
| |
| def print_reply(self, req, reply): |
| print("Reply %s to %s" % (reply.getlayer(IP).dst,reply.dst)) |
| |
| def make_reply(self, req): |
| mac = req.src |
| if isinstance(self.pool, list): |
| if mac not in self.leases: |
| self.leases[mac] = self.pool.pop() |
| ip = self.leases[mac] |
| else: |
| ip = self.pool |
| |
| repb = req.getlayer(BOOTP).copy() |
| repb.op="BOOTREPLY" |
| repb.yiaddr = ip |
| repb.siaddr = self.gw |
| repb.ciaddr = self.gw |
| repb.giaddr = self.gw |
| del(repb.payload) |
| rep=Ether(dst=mac)/IP(dst=ip)/UDP(sport=req.dport,dport=req.sport)/repb |
| return rep |
| |
| |
| class DHCP_am(BOOTP_am): |
| function_name="dhcpd" |
| def make_reply(self, req): |
| resp = BOOTP_am.make_reply(self, req) |
| if DHCP in req: |
| dhcp_options = [(op[0],{1:2,3:5}.get(op[1],op[1])) |
| for op in req[DHCP].options |
| if isinstance(op, tuple) and op[0] == "message-type"] |
| dhcp_options += [("server_id",self.gw), |
| ("domain", self.domain), |
| ("router", self.gw), |
| ("name_server", self.gw), |
| ("broadcast_address", self.broadcast), |
| ("subnet_mask", self.netmask), |
| ("renewal_time", self.renewal_time), |
| ("lease_time", self.lease_time), |
| "end" |
| ] |
| resp /= DHCP(options=dhcp_options) |
| return resp |
| |
| |
