Skip to content

Instantly share code, notes, and snippets.

@idkiller

idkiller/dns_sniff.py

Last active May 12, 2020 04:00
Show Gist options
  • Save idkiller/b9a8b0b83a69c3cefc5a3369cc655f98 to your computer and use it in GitHub Desktop.
Save idkiller/b9a8b0b83a69c3cefc5a3369cc655f98 to your computer and use it in GitHub Desktop.
Download ZIP
python mDNS / DNS-SD sniff tool (Parsing DNS-SD packet!)
Raw
dns_sniff.py
# -*- coding: utf-8 -*-
import socket
import struct
from time import gmtime, strftime
import re
#import binascii
import string
class packetSniff():
def __init__(self):
self.hostip = '224.0.0.251'
def analDNSPacket(self, packet):
now = strftime("%Y-%M-%d %H:%M:%S", gmtime())
recvHeader = struct.unpack_from('!HHHHHH', packet[:12])
recvData = packet[12:].split(b'\x00', 1)
flag_QnA = recvHeader[1] & 0b1000000000000000
queryType, queryClass = struct.unpack('!HH', recvData[1][:4])
rData = packet[12:]
name = self.parseDomainName(rData[:rData.index(b'\x00')], packet)
#if not name.startswith('._jenkins'):
# return
if flag_QnA == 0 :
print('[DNS Query]------------------------------------')
else :
print('[DNS Response]---------------------------------')
print('[*] Time : ' + now)
print('[+] Transaction ID : ' + str(hex(recvHeader[0])))
print('[+] Flags : ' + str(hex(recvHeader[1])))
print('[+] Number of queries : ' + str(hex(recvHeader[2])))
print('[+] Number of authoritative : ' + str(hex(recvHeader[3])))
print('[+] Number of additional record : ' + str(hex(recvHeader[4])))
print('[+] Query Name : ' + name)
print('[+] Query Type : ' + str(hex(queryType))) + '('+str(queryType)+')'
print('[+] Query Class : ' + str(hex(queryClass)))
if flag_QnA != 0:
ttl, rdlength = struct.unpack('!IH', recvData[1][4:10])
print('[+] TTL : ' + str(ttl))
print('[+] RDLength : ' + str(rdlength))
rdata = recvData[1][10: 10 + rdlength]
print('[+] DATA : ')
if queryType == 16: # TXT
index = 0
while index < len(rdata):
length = ord(rdata[index])
data = rdata[index+1:index+1+length]
print(' ' + data)
index += length + 1
elif queryType == 1: # A
print(' Address : ' + '.'.join([str(ord(x)) for x in rdata]) )
elif queryType == 28: # AAAA
addr = list(struct.unpack('!HHHHHHHH', rdata))
print(' Address : ' + '.'.join([str(a) for a in addr]))
elif queryType == 33: # SRV
priority, weight, port, length = struct.unpack('!HHHB', rdata[:7])
print(' Priority : ' + str(priority))
print(' Weight : ' + str(weight))
print(' Port : ' + str(port))
print(' Length : ' + str(length))
#print(' Target : ' + rdata[7:length+7])
print(' Target : ' + self.parseDomainName(rdata[7:], packet))
elif queryType == 12: # PTR
print(' Domain : ' + self.parseDomainName(rdata, packet))
else:
print ' '.join('{:02x}'.format(ord(x)) for x in rdata)
print ' '.join('{:>2}'.format(x) if x in string.printable else '□ ' for x in rdata)
print('-----------------------------------------------')
def parseLabel(self, data):
chars = []
i = 0
size = 0
while i < len(data) and data[i] != b'\x00':
size = ord(data[i])
chars.append(data[i+1:i+1+size].encode('utf-8'))
i += size + 1
return '.'.join(chars)
def parseDomainName(self, rdata, packet):
if ord(rdata[-2]) & 0xc0 == 0xc0:
index = (ord(rdata[-2]) & 0x3f) << 8 | ord(rdata[-1])
return self.parseLabel(rdata[:-2]) + '.' + self.parseLabel(packet[index:])
else:
return self.parseLabel(rdata)
def start(self):
rawsock = socket.socket(family=socket.AF_INET, type=socket.SOCK_RAW, proto=17)
rawsock.bind((self.hostip, 0))
# rawsock.ioctl(socket.SIO_RCVALL, socket.RCVALL_ON)
print('[*] Start Sniffer Process!')
while(True):
packet, address = rawsock.recvfrom(65565)
srcport = struct.unpack_from('!H', packet[20:22])[0]
dstport = struct.unpack_from('!H', packet[22:24])[0]
if(srcport == 5353 or dstport == 5353):
now = strftime("%H:%M:%S", gmtime())
self.analDNSPacket(packet[28:])
if __name__ == '__main__':
obj = packetSniff()
obj.start()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment