# Impacket - Collection of Python classes for working with network protocols. # # Copyright Fortra, LLC and its affiliated companies # # All rights reserved. # # This software is provided under a slightly modified version # of the Apache Software License. See the accompanying LICENSE file # for more information. # # Description: # RFCs for the DNS Server service # # - 1034 - Domain Names -- Concepts and Facilities [https://www.ietf.org/rfc/rfc1034.txt] # - 1035 - Domain Names -- Implementation and Specification [https://www.ietf.org/rfc/rfc1035.txt] # - 1123 - Requirements for Internet Hosts -- Application and Support [https://www.ietf.org/rfc/rfc1123.txt] # - 1886 - DNS Extensions to Support IP Version 6 [https://www.ietf.org/rfc/rfc1886.txt] # - 1995 - Incremental Zone Transfer in DNS [https://www.ietf.org/rfc/rfc1995.txt] # - 1996 - A Mechanism for Prompt Notification of Zone Changes (DNS NOTIFY) [https://www.ietf.org/rfc/rfc1996.txt] # - 2136 - Dynamic Updates in the Domain Name System (DNS UPDATE) [https://www.ietf.org/rfc/rfc2136.txt] # - 2181 - Clarifications to the DNS Specification [https://www.ietf.org/rfc/rfc2181.txt] # - 2308 - Negative Caching of DNS Queries (DNS NCACHE) [https://www.ietf.org/rfc/rfc2308.txt] # - 2535 - Domain Name System Security Extensions (DNSSEC) [https://www.ietf.org/rfc/rfc2535.txt] # - 2671 - Extension Mechanisms for DNS (EDNS0) [https://www.ietf.org/rfc/rfc2671.txt] # - 2782 - A DNS RR for specifying the location of services (DNS SRV) [https://www.ietf.org/rfc/rfc2782.txt] # - 2930 - Secret Key Establishment for DNS (TKEY RR) [https://www.ietf.org/rfc/rfc2930.txt] # - 3645 - Generic Security Service Algorithm for Secret Key Transaction Authentication for DNS (GSS-TSIG) [https://www.ietf.org/rfc/rfc3645.txt] # - 3646 - DNS Configuration options for Dynamic Host Configuration Protocol for IPv6 (DHCPv6) [https://www.ietf.org/rfc/rfc3646.txt] # # Author: # Andres Blanco # Gustavo Moreira # import socket import struct from impacket.ImpactPacket import ProtocolPacket class DNSFlags(): 'Bitmap with the flags of a dns packet.' # QR - Query/Response - 1 bit QR_QUERY = int("0000000000000000", 2) QR_RESPONSE = int("1000000000000000", 2) # OP - Opcode - 4 bits OP_STANDARD_QUERY = int("0000000000000000", 2) # Standard query. OP_INVERSE_QUERY = int("0100000000000000", 2) # Inverse query. OP_STATUS_QUERY = int("0010000000000000", 2) # Server status request. OP_NOTIFY = int("0000100000000000", 2) # Notify. OP_UPDATE = int("0100100000000000", 2) # Update. # AA - Authority Answer - 1 bit AA_NOT_AUTH_ANSWER = int("0000000000000000", 2) # Not authoritative. AA_AUTH_ANSWER = int("0000010000000000", 2) # Is authoritative. # TC - Truncated - 1 bit TC_NOT_TRUNCATED = int("0000000000000000", 2) # Not truncated. TC_TRUNCATED = int("0000001000000000", 2) # Message truncated. # RD - Recursion Desired - 1 bit RD_NOT_RECURSIVE_QUERY = int("0000000000000000", 2) # Recursion not desired. RD_RECURSIVE_QUERY = int("0000000100000000", 2) # Recursion desired. # RA - Recursion Available - 1 bit RA_NOT_AVAILABLE = int("0000000000000000", 2) # Recursive query support not available. RA_AVAILABLE = int("0000000010000000", 2) # Recursive query support available. # Z - 3 bits Z = int("0000000000000000", 2) # AD - Authenticated Data - 1 bit AUTHENTICATED_DATA = int("0000000000100000", 2) # CD - Checking Disabled - 1 bit CHECKING_DISABLED = int("0000000000010000", 2) # RCODE - 4 bits RCODE_NO_ERROR = int("0000000000000000", 2) # The request completed successfully. RCODE_FORMAT_ERROR = int("0000000000001000", 2) # The name server was unable to interpret the query. RCODE_SERVER_FAILURE = int("0000000000000100", 2) # The name server was unable to process this query due to a problem with the name server. RCODE_NAME_ERROR = int("0000000000001100", 2) # Meaningful only for responses from an authoritative name server, this code signifies that the domain name referenced in the query does not exist. RCODE_NOT_IMPLEMENTED = int("0000000000000010", 2) # Not Implemented. The name server does not support the requested kind of query. RCODE_REFUSED = int("0000000000001010", 2) # The name server refuses to perform the specified operation for policy reasons. RCODE_YXDOMAIN = int("0000000000000110", 2) # Name Exists when it should not. RCODE_YXRRSET = int("0000000000001110", 2) # RR Set Exists when it should not. RCODE_NXRRSET = int("0000000000000001", 2) # RR Set that should exist does not. RCODE_NOAUTH = int("0000000000001001", 2) # Server Not Authoritative for zone. RCODE_NOTZONE = int("0000000000000101", 2) # Name not contained in zone. class DNSType(): A = 1 # IPv4 address. NS = 2 # Authoritative name server. MD = 3 # Mail destination. Obsolete use MX instead. MF = 4 # Mail forwarder. Obsolete use MX instead. CNAME = 5 # Canonical name for an alias. SOA = 6 # Marks the start of a zone of authority. MB = 7 # Mailbox domain name. MG = 8 # Mail group member. MR = 9 # Mail rename domain name. NULL = 10 # Null resource record. WKS = 11 # Well known service description. PTR = 12 # Domain name pointer. HINFO = 13 # Host information. MINFO = 14 # Mailbox or mail list information. MX = 15 # Mail exchange. TXT = 16 # Text strings. RP = 17 # Responsible Person. AFSDB = 18 # AFS Data Base location. X25 = 19 # X.25 PSDN address. ISDN = 20 # ISDN address. RT = 21 # Route Through. NSAP = 22 # NSAP address. NSAP style A record. NSAP_PTR = 23 # NSAP pointer. SIG = 24 # Security signature. KEY = 25 # Security key. PX = 26 # X.400 mail mapping information. GPOS = 27 # Geographical Position. AAAA = 28 # IPv6 Address. LOC = 29 # Location Information. NXT = 30 # Next Domain (obsolete). EID = 31 # Endpoint Identifier. NB = 32 # NetBIOS general Name Service. NBSTAT = 33 # NetBIOS NODE STATUS. ATMA = 34 # ATM Address. NAPTR = 35 # Naming Authority Pointer. KX = 36 # Key Exchanger. CERT = 37 A6 = 38 DNAME = 39 SINK = 40 OPT = 41 APL = 42 DS = 43 # Delegation Signer. SSHFP = 44 # SSH Key Fingerprint. IPSECKEY = 45 RRSIG = 46 NSEC = 47 # NextSECure. DNSKEY = 48 DHCID = 49 # DHCP identifier. NSEC3 = 50 NSEC3PARAM = 51 HIP = 55 # Host Identity Protocol. NINFO = 56 RKEY = 57 SPF = 99 # Sender Policy Framework. UINFO = 100 UID = 101 GID = 102 UNSPEC = 103 TKEY = 249 TSIG = 250 # Transaction Signature. IXFR = 251 # Incremental transfer. AXFR = 252 # A request for a transfer of an entire zone. MAILB = 253 # A request for mailbox-related records (MB, MG or MR). MAILA = 254 # A request for mail agent RRs. Obsolete. ALL = 255 # A request for all records. DNSSEC = 32768 # Trust Authorities. DNSSEC = 32769 # DNSSEC Lookaside Validation. @staticmethod def getTypeName(type): for item, value in list(DNSType.__dict__.items()): if value == type: return item class DNSClass(): RESERVED = 0 IN = 1 # Internet. CH = 3 # Chaos. HS = 4 # Hesiod. NONE = 254 ANY = 255 # QCLASS only @staticmethod def getClassName(type): for item, value in list(DNSClass.__dict__.items()): if value == type: return item class DNS(ProtocolPacket): '''The Message Header is present in all messages. Never empty. Contains various flags and values which control the transaction.''' __TYPE_LEN = 2 # Unsigned 16 bit value. __CLASS_LEN = 2 # Unsigned 16 bit value. __POINTER_LEN = 2 # A pointer is an unsigned 16-bit value. __TTL_LEN = 4 # Unsigned 32 bit value. The time in seconds that the record may be cached. __RDLENGTH_LEN = 2 # Unsigned 16-bit value that defines the length in bytes (octets) of the RDATA record. __TYPE_A_LEN = 4 # Unsigned 32-bit value representing the IP address. __SERIAL_LEN = 4 # Serial Number Unsigned 32-bit integer. __REFRESH_LEN = 4 # Refresh interval Unsigned 32-bit integer. __RETRY_LEN = 4 # Retry Interval Unsigned 32-bit integer. __EXPIRATION_LEN = 4 # Expiration Limit Unsigned 32-bit integer. __MINTTL_LEN = 4 # Minimum TTL Unsigned 32-bit integer. __PREF_LEN = 2 # Preference Unsigned 16-bit integer. __IS_POINTER = int("11000000", 2) __OFFSETMASK = int("00111111", 2) def __init__(self, aBuffer = None): self.__HEADER_BASE_SIZE = 12 self.__TAIL_SIZE = 0 ProtocolPacket.__init__(self, self.__HEADER_BASE_SIZE, self.__TAIL_SIZE) if aBuffer: self.load_packet(aBuffer) def get_transaction_id(self): 'Get 16 bit message ID.' return self.header.get_word(0) def set_transaction_id(self, value): 'Set 16 bit message ID.' self.header.set_word(0, value) def get_transaction_id_tcp(self): 'Get 16 bit message ID.' return self.header.get_word(2) def set_transaction_id_tcp(self, value): 'Set 16 bit message ID.' self.header.set_word(2, value) def get_flags(self): 'Get 16 bit flags.' return self.header.get_word(2) def set_flags(self, value): 'Set 16 bit flags.' self.header.set_word(2, value) def get_flags_tcp(self): 'Get 16 bit flags.' return self.header.get_word(4) def set_flags_tcp(self, value): 'Set 16 bit flags.' self.header.set_word(4, value) def get_qdcount(self): 'Get Unsigned 16 bit integer specifying the number of entries in the question section.' return self.header.get_word(4) def set_qdcount(self, value): 'Set Unsigned 16 bit integer specifying the number of entries in the question section.' self.header.set_word(4, value) def get_qdcount_tcp(self): 'Get Unsigned 16 bit integer specifying the number of entries in the question section.' return self.header.get_word(6) def set_qdcount_tcp(self, value): 'Set Unsigned 16 bit integer specifying the number of entries in the question section.' self.header.set_word(6, value) def get_ancount(self): 'Get Unsigned 16 bit integer specifying the number of resource records in the answer section' return self.header.get_word(6) def set_ancount(self, value): 'Set Unsigned 16 bit integer specifying the number of resource records in the answer section' self.header.set_word(6, value) def get_nscount(self): 'Get Unsigned 16 bit integer specifying the number of name server resource records in the authority section.' return self.header.get_word(8) def set_nscount(self, value): 'Set Unsigned 16 bit integer specifying the number of name server resource records in the authority section.' self.header.set_word(8, value) def get_arcount(self): 'Get Unsigned 16 bit integer specifying the number of resource records in the additional records section.' return self.header.get_word(10) def set_arcount(self, value): 'Set Unsigned 16 bit integer specifying the number of resource records in the additional records section.' self.header.set_word(10, value) def get_questions(self): 'Get a list of the DNS Question.' return self.__get_questions()[0] def __get_questions(self): aux = [] offset = 0 qdcount = self.get_qdcount() data = self.get_body_as_string() for _ in range(qdcount): # number of questions offset, qname = self.parseCompressedMessage(data, offset) qtype = data[offset:offset+self.__TYPE_LEN] offset += self.__TYPE_LEN qclass = data[offset:offset+self.__CLASS_LEN] offset += self.__CLASS_LEN qtype = struct.unpack("!H", qtype)[0] qclass = struct.unpack("!H", qclass)[0] aux.append((qname, qtype, qclass)) return (aux, offset) def get_questions_tcp(self): 'Get a list of the DNS Question.' return self.__get_questions_tcp()[0] def __get_questions_tcp(self): aux = [] offset = 2 qdcount = self.get_qdcount_tcp() data = self.get_body_as_string() for _ in range(qdcount): # number of questions offset, qname = self.parseCompressedMessage(data, offset) qtype = data[offset:offset+self.__TYPE_LEN] offset += self.__TYPE_LEN qclass = data[offset:offset+self.__CLASS_LEN] offset += self.__CLASS_LEN qtype = struct.unpack("!H", qtype)[0] qclass = struct.unpack("!H", qclass)[0] aux.append((qname, qtype, qclass)) return (aux, offset) def parseCompressedMessage(self, buf, offset=0): 'Parse compressed message defined on rfc1035 4.1.4.' if offset >= len(buf): raise Exception("No more data to parse. Offset is bigger than length of buffer.") byte = struct.unpack("B", buf[offset:offset+1])[0] # if the first two bits are ones (11000000=0xC0), the next bits are the offset if byte & 0xC0 == 0xC0: # It's a pointer pointer = struct.unpack("!H", buf[offset:offset+2])[0] # network unsigned short pointer = (pointer & 0x3FFF) - self.__HEADER_BASE_SIZE if offset == pointer: raise Exception("The infinite loop is in DNS decompression. Encountered pointer points to the current offset.") offset += 2 name = self.parseCompressedMessage(buf, pointer)[1] return (offset, name) else: # It's a label if byte == 0x00: offset += 1 return (offset, '') offset += 1 name = buf[offset:offset+byte] offset += byte offset, unnamed = self.parseCompressedMessage(buf, offset) if not unnamed: return (offset, name) else: return (offset, name + b"." + unnamed) def get_answers(self): return self.__get_answers()[0] def get_authoritative(self): return self.__get_authoritative()[0] def get_additionals(self): return self.__get_additionals()[0] def __get_answers(self): offset = self.__get_questions()[1] # get the initial offset ancount = self.get_ancount() return self.__process_answer_structure(offset, ancount) def __get_authoritative(self): 'Get a list of the DNS Authoritative.' offset = self.__get_answers()[1] # get the initial offset nscount = self.get_nscount() return self.__process_answer_structure(offset, nscount) def __get_additionals(self): 'Get a list of the DNS Additional Records.' offset = self.__get_authoritative()[1] # get the initial offset arcount = self.get_arcount() return self.__process_answer_structure(offset, arcount) def __process_answer_structure(self, offset, num): aux = [] data = self.get_body_as_string() for _ in range(num): offset, qname = self.parseCompressedMessage(data, offset) qtype = data[offset:offset+self.__TYPE_LEN] qtype = struct.unpack("!H", qtype)[0] offset += self.__TYPE_LEN qclass = data[offset:offset+self.__CLASS_LEN] qclass = struct.unpack("!H", qclass)[0] offset += self.__CLASS_LEN qttl_raw = data[offset:offset+self.__TTL_LEN] qttl = struct.unpack("!L", qttl_raw)[0] offset += self.__TTL_LEN qrdlength = data[offset:offset+self.__RDLENGTH_LEN] qrdlength = struct.unpack("!H", qrdlength)[0] offset += self.__RDLENGTH_LEN qrdata = {} if qtype == DNSType.A: # IP Address Unsigned 32-bit value representing the IP address qrdata["IPAddress"] = socket.inet_ntoa(data[offset:offset+qrdlength]) offset += self.__TYPE_A_LEN elif qtype == DNSType.SOA: # Primary NS Variable length. The name of the Primary Master for the domain. May be a label, pointer or any combination. offset, primaryNs = self.parseCompressedMessage(data, offset) qrdata["PrimaryNS"] = primaryNs # Admin MB Variable length. The administrator's mailbox. May be a label, pointer or any combination. offset, adminMb = self.parseCompressedMessage(data, offset) qrdata["AdminMB"] = adminMb # Serial Number Unsigned 32-bit integer. qrdata["SerialNumber"] = struct.unpack("!L", data[offset:offset+self.__SERIAL_LEN])[0] offset += self.__SERIAL_LEN # Refresh interval Unsigned 32-bit integer. qrdata["RefreshInterval"] = struct.unpack("!L", data[offset:offset+self.__REFRESH_LEN])[0] offset += self.__REFRESH_LEN # Retry Interval Unsigned 32-bit integer. qrdata["RetryInterval"] = struct.unpack("!L", data[offset:offset+self.__RETRY_LEN])[0] offset += self.__RETRY_LEN # Expiration Limit Unsigned 32-bit integer. qrdata["ExpirationLimit"] = struct.unpack("!L", data[offset:offset+self.__EXPIRATION_LEN])[0] offset += self.__EXPIRATION_LEN # Minimum TTL Unsigned 32-bit integer. qrdata["MinimumTTL"] = struct.unpack("!L", data[offset:offset+self.__MINTTL_LEN])[0] offset += self.__MINTTL_LEN elif qtype == DNSType.MX: # Preference Unsigned 16-bit integer. qrdata["Preference"] = struct.unpack("!H", data[offset:offset+self.__PREF_LEN])[0] # Mail Exchanger The name host name that provides the service. May be a label, pointer or any combination. offset, mailExch = self.parseCompressedMessage(data, offset) qrdata["MailExchanger"] = mailExch elif qtype == DNSType.PTR or qtype == DNSType.NS or qtype == DNSType.CNAME: # Name The host name that represents the supplied IP address (in the case of a PTR) or the NS name for the supplied domain (in the case of NS). May be a label, pointer or any combination. offset, name = self.parseCompressedMessage(data, offset) qrdata["Name"] = str(name.decode('ascii')) elif qtype == DNSType.OPT: # rfc2671 4.3 #NAME domain name empty (root domain) #TYPE u_int16_t OPT #CLASS u_int16_t sender's UDP payload size #TTL u_int32_t extended RCODE and flags #RDLEN u_int16_t describes RDATA #RDATA octet stream {attribute,value} pairs #udp_payload = qclass udp_payload_size = qclass ext_rcode = struct.unpack("B", qttl_raw[0:1])[0] version = struct.unpack("B", qttl_raw[1:2])[0] flags = struct.unpack("!H", qttl_raw[2:4])[0] qrdata["RDATA"] = data[offset:offset+qrdlength] offset += qrdlength aux.append((qname, qtype, udp_payload_size, ext_rcode, version, flags, qrdata)) continue else: # We don't know how to parse it, just skip it offset += qrdlength aux.append((qname, qtype, qclass, qttl, qrdata)) return (aux, offset) def get_header_size(self): return self.__HEADER_BASE_SIZE def __str__(self): res = "" id = self.get_transaction_id() flags = self.get_flags() qdcount = self.get_qdcount() ancount = self.get_ancount() nscount = self.get_nscount() arcount = self.get_arcount() res += "DNS " if flags & DNSFlags.QR_RESPONSE: res += "RESPONSE\n" else: res += "QUERY\n" res += " - Transaction ID -- [0x%04x] %d\n" % (id, id) res += " - Flags ----------- [0x%04x] %d\n" % (flags, flags) res += " - QdCount --------- [0x%04x] %d\n" % (qdcount, qdcount) res += " - AnCount --------- [0x%04x] %d\n" % (ancount, ancount) res += " - NsCount --------- [0x%04x] %d\n" % (nscount, nscount) res += " - ArCount --------- [0x%04x] %d\n" % (arcount, arcount) if qdcount > 0: res += " - Questions:\n" questions = self.get_questions() questions.reverse() while(questions): qname, qtype, qclass = questions.pop() format = (str(qname.decode('ascii')), DNSType.getTypeName(qtype), qtype, DNSClass.getClassName(qclass), qclass) res += " * Domain: %s - Type: %s [0x%04x] - Class: %s [0x%04x]\n" % format if ancount > 0: res += " - Answers:\n" answers = self.get_answers() answers.reverse() while(answers): qname, qtype, qclass, qttl, qrdata = answers.pop() format = (str(qname.decode('ascii')), DNSType.getTypeName(qtype), qtype, DNSClass.getClassName(qclass), qclass, qttl, repr(qrdata)) res += " * Domain: %s - Type: %s [0x%04x] - Class: %s [0x%04x] - TTL: %d seconds - %s\n" % format if nscount > 0: res += " - Authoritative:\n" authoritative = self.get_authoritative() authoritative.reverse() while(authoritative): qname, qtype, qclass, qttl, qrdata = authoritative.pop() format = (str(qname.decode('ascii')), DNSType.getTypeName(qtype), qtype, DNSClass.getClassName(qclass), qclass, qttl, repr(qrdata)) res += " * Domain: %s - Type: %s [0x%04x] - Class: %s [0x%04x] - TTL: %d seconds - %s\n" % format if arcount > 0: res += " - Additionals:\n" additionals = self.get_additionals() for additional in additionals: qtype = additional[1] if qtype == DNSType.OPT: qname, qtype, udp_payload_size, ext_rcode, version, flags, qrdata = additional format = (DNSType.getTypeName(qtype), qtype, udp_payload_size, ext_rcode, version, flags, repr(qrdata['RDATA'])) res += " * Name: - Type: %s [0x%04x] - udp payload size: [%d] - extended RCODE: [0x%02x] - EDNS0 version: [0x%02x] - Z Flags: [0x%02x] - RDATA: [%s]\n" % format else: qname, qtype, qclass, qttl, qrdata = additional format = (str(qname.decode('ascii')), DNSType.getTypeName(qtype), qtype, DNSClass.getClassName(qclass), qclass, qttl, repr(qrdata)) res += " * Domain: %s - Type: %s [0x%04x] - Class: %s [0x%04x] - TTL: %d seconds - %s\n" % format return res def __get_questions_raw(self): if self.get_qdcount() == 0: return '' questions_offset = self.__get_questions()[1] raw_data = self.get_body_as_string()[:questions_offset] return raw_data def __get_answers_raw(self): if self.get_ancount() == 0: return '' questions_offset = self.__get_questions()[1] answers_offset = self.__get_answers()[1] raw_data = self.get_body_as_string()[questions_offset: answers_offset] return raw_data def __get_authoritative_raw(self): if self.get_nscount() == 0: return '' answers_offset = self.__get_answers()[1] authoritative_offset = self.__get_authoritative()[1] raw_data = self.get_body_as_string()[answers_offset:authoritative_offset] return raw_data def __get_additionals_raw(self): if self.get_arcount() == 0: return '' authoritative_offset = self.__get_authoritative()[1] raw_data = self.get_body_as_string()[authoritative_offset:] return raw_data def add_answer(self, answer_raw): '''Add a raw answer''' questions_raw = self.__get_questions_raw() answers_raw = self.__get_answers_raw() authoritative_raw = self.__get_authoritative_raw() additionals_raw = self.__get_additionals_raw() answers_raw += answer_raw body = questions_raw + answers_raw + authoritative_raw + additionals_raw self.load_body(body) # It breaks children hierarchy # Increment the answer count cur_answer_count = self.get_ancount()+1 self.set_ancount(cur_answer_count) def is_edns0(self): additionals = self.get_additionals() for item in additionals: response_type = item[1] if response_type == DNSType.OPT: return True return False