How to use the dpkt.dns.DNS function in dpkt
To help you get started, we’ve selected a few dpkt examples, based on popular ways it is used in public projects.
Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
stamparm / maltrail / maltrail.py View on Github 
if url in _blacklists[BLACKLIST.URL]:
src, dst, type_, trail, info, reference = src_ip, dst_ip, BLACKLIST.URL, url, _blacklists[BLACKLIST.URL][url][0], _blacklists[BLACKLIST.URL][url][1]
store_db(sec, usec, src, dst, type_, trail, info, reference)
elif protocol == socket.IPPROTO_UDP:
_check_ips()
i = iph_length + ETH_LENGTH
src_port, dst_port = struct.unpack("!HH", packet[i:i + 4])
if dst_port == 53:
h_size = ETH_LENGTH + iph_length + 8
data = packet[h_size:]
try:
dns = dpkt.dns.DNS(data)
except:
pass
else:
if dns.opcode == dpkt.dns.DNS_QUERY:
for query in dns.qd:
domain = query.name
parts = domain.split('.')
for i in xrange(0, len(parts) - 1):
_ = '.'.join(parts[i:])
if _ in _blacklists[BLACKLIST.DNS]:
src, dst, type_, trail, info, reference = src_ip, dst_ip, BLACKLIST.DNS, domain, _blacklists[BLACKLIST.DNS][_][0], _blacklists[BLACKLIST.DNS][_][1]
store_db(sec, usec, src, dst, type_, trail, info, reference)
break
except struct.error:
passtry:
#Some times buf contains \x00 so this checks skip them
cnt +=1
if buf == '\x00\x00\x00\x00':
#Some pcap files have bytes 00 in beginning, just discard them
continue
else:
eth = dpkt.ethernet.Ethernet(buf)
ip = eth.data
tcp = ip.data
if tcp.__class__.__name__ in protocols:
if tcp.sport == 53 or tcp.dport == 53:
if eth.type == 2048 and ip.p == 17 :
try:
dns = dpkt.dns.DNS(tcp.data)
except:
#print "Error:DNS"
continue # Discard errornous Data
if dns.qr == dpkt.dns.DNS_R and dns.opcode == dpkt.dns.DNS_QUERY and dns.rcode == dpkt.dns.DNS_RCODE_NOERR:
if len(dns.an) >= 1:
for answer in dns.an:
req = { }
if answer.type == 1: #DNS_A
req = {'type': 'A', 'name': unicode(answer.name, errors='replace'),'result': socket.inet_ntoa(answer.rdata)}
elif answer.type == 5: # "CNAME request"
req = {'type': 'CN', 'name': unicode(answer.name, errors='replace'),'result': unicode(answer.cname, errors='replace')}
elif answer.type == 12:
#print "PTR request", answer.name, "\tresponse", answer.ptrn
req = {'type': 'PTR', 'name': unicode(answer.name, errors='replace'),'result': unicode(answer.ptrname, errors='replace')}
if req <> {} : dns_list.append(req)
except Exception as e:for (src, sport, dst, dport, data ) in udp_iterator(pc) :
# Uncomment if you want to see all UDP packets
# print "from ", socket.inet_ntoa(src),":",sport, " to ", socket.inet_ntoa(dst),":",dport
if dport == 53 :
# UDP/53 is a DNS query
dns = dpkt.dns.DNS(data)
if dns.opcode != dpkt.dns.DNS_QUERY :
print "A DNS packet was sent to the nameserver, but the opcode was %d instead of DNS_QUERY (this is a software error)" % dns.opcode
if dns.qr != dpkt.dns.DNS_Q :
print "A DNS packet was sent to the name server, but dns.qr is not 0 and should be. It is %d" % dns.qr
print "query for ", dns.qd[0].name, "ID is ", dns.id, "dns.qr is ", dns.qr, "query type is ", dns.qd[0].type, type_table[dns.qd[0].type]
print "dns.qd is ", dns.qd
elif sport == 53 :
# UDP/53 is a DNS response
dns = dpkt.dns.DNS(data)
print "responding to ", dns.id, "dns.qr is ", dns.qr
if dns.qr != dpkt.dns.DNS_R :
print "A DNS packet was received from a name server, but dns.qr is not 1 and should be. It is %d" % dns.qr
if dns.get_rcode() == dpkt.dns.DNS_RCODE_NOERR :
print "Response has no error"
elif dns.get_rcode() == dpkt.dns.DNS_RCODE_NXDOMAIN :
print "There is no name in this domain"
else :
print "Response is something other than NOERR or NXDOMAIN %d - this software is incomplete" % dns.get_rcode()
print "The response packet has %d RRs" % len(dns.an)
# Decode the RR records in the NS section
for rr in dns.ns :
decode_dns_response ( rr, "NS")
# Decode the answers in the DNS answer
for rr in dns.an :
decode_dns_response ( rr, "AN" )def test_basic():
from . import ip
s = b'E\x00\x02\x08\xc15\x00\x00\x80\x11\x92aBk0\x01Bk0w\x005\xc07\x01\xf4\xda\xc2d\xd2\x81\x80\x00\x01\x00\x03\x00\x0b\x00\x0b\x03www\x06google\x03com\x00\x00\x01\x00\x01\xc0\x0c\x00\x05\x00\x01\x00\x00\x03V\x00\x17\x03www\x06google\x06akadns\x03net\x00\xc0,\x00\x01\x00\x01\x00\x00\x01\xa3\x00\x04@\xe9\xabh\xc0,\x00\x01\x00\x01\x00\x00\x01\xa3\x00\x04@\xe9\xabc\xc07\x00\x02\x00\x01\x00\x00KG\x00\x0c\x04usw5\x04akam\xc0>\xc07\x00\x02\x00\x01\x00\x00KG\x00\x07\x04usw6\xc0t\xc07\x00\x02\x00\x01\x00\x00KG\x00\x07\x04usw7\xc0t\xc07\x00\x02\x00\x01\x00\x00KG\x00\x08\x05asia3\xc0t\xc07\x00\x02\x00\x01\x00\x00KG\x00\x05\x02za\xc07\xc07\x00\x02\x00\x01\x00\x00KG\x00\x0f\x02zc\x06akadns\x03org\x00\xc07\x00\x02\x00\x01\x00\x00KG\x00\x05\x02zf\xc07\xc07\x00\x02\x00\x01\x00\x00KG\x00\x05\x02zh\xc0\xd5\xc07\x00\x02\x00\x01\x00\x00KG\x00\x07\x04eur3\xc0t\xc07\x00\x02\x00\x01\x00\x00KG\x00\x07\x04use2\xc0t\xc07\x00\x02\x00\x01\x00\x00KG\x00\x07\x04use4\xc0t\xc0\xc1\x00\x01\x00\x01\x00\x00\xfb4\x00\x04\xd0\xb9\x84\xb0\xc0\xd2\x00\x01\x00\x01\x00\x001\x0c\x00\x04?\xf1\xc76\xc0\xed\x00\x01\x00\x01\x00\x00\xfb4\x00\x04?\xd7\xc6S\xc0\xfe\x00\x01\x00\x01\x00\x001\x0c\x00\x04?\xd00.\xc1\x0f\x00\x01\x00\x01\x00\x00\n\xdf\x00\x04\xc1-\x01g\xc1"\x00\x01\x00\x01\x00\x00\x101\x00\x04?\xd1\xaa\x88\xc15\x00\x01\x00\x01\x00\x00\r\x1a\x00\x04PCC\xb6\xc0o\x00\x01\x00\x01\x00\x00\x10\x7f\x00\x04?\xf1I\xd6\xc0\x87\x00\x01\x00\x01\x00\x00\n\xdf\x00\x04\xce\x84dl\xc0\x9a\x00\x01\x00\x01\x00\x00\n\xdf\x00\x04A\xcb\xea\x1b\xc0\xad\x00\x01\x00\x01\x00\x00\x0b)\x00\x04\xc1l\x9a\t'
ip = ip.IP(s)
my_dns = DNS(ip.udp.data)
assert my_dns.qd[0].name == 'www.google.com' and my_dns.an[1].name == 'www.google.akadns.net'
s = b'\x05\xf5\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00\x03www\x03cnn\x03com\x00\x00\x01\x00\x01'
my_dns = DNS(s)
assert s == bytes(my_dns)def blobHandler(self, conn, blob):
'''for each blob, examine each segment (UDP packet) seperately as each will be a DNS Q/A
pair Q/A by ID and return as pairs'''
connrqs = self.requests.setdefault(conn, {})
# iterate blob as each packet will be a seperate request (catches spoofing)
for data in blob:
try:
dns = dpkt.dns.DNS(data)
except Exception, e:
self._exc(e)
continue
if dns.qr == dpkt.dns.DNS_Q:
connrqs[dns.id] = [blob.starttime, dns, 0]
elif dns.qr == dpkt.dns.DNS_A:
rq = connrqs.get(dns.id, [None, None, 0])
rq[2] += 1
if "DNSHandler" in dir(self):
self.DNSHandler(conn=conn, request=rq[1], response=dns, requesttime=rq[0],
responsetime=blob.starttime, responsecount=rq[2])def _add_dns(self, conn, udpdata):
"""Add a DNS data flow.
@param udpdata: UDP data flow.
"""
dns = dpkt.dns.DNS(udpdata)
if dns.qr == dpkt.dns.DNS_Q:
self._add_used_dns_server(conn["dst"])
# DNS query parsing.
query = {}
# Temporary list for found A or AAAA responses.
_ip = []
if dns.rcode == dpkt.dns.DNS_RCODE_NOERR or \
dns.qr == dpkt.dns.DNS_R or \
dns.opcode == dpkt.dns.DNS_QUERY or True:
# DNS question.
try:
q_name = dns.qd[0].name
q_type = dns.qd[0].typetcp_dport))
rtn_list.append({'protocol':'TCP',
'ip_src': ip_src,
'ip_dst': ip_dst,
'port_src': tcp_sport,
'port_dst': tcp_dport})
# UDP
elif ip.p == 0x11:
udp = ip.data
udp_sport = udp['sport']
udp_dport = udp['dport']
dns_hosts = []
if udp_sport == '53' or udp_dport == 53:
dns = dpkt.dns.DNS(udp.data)
for q in dns['qd']:
dns_hosts.append(q['name'])
logger.debug("UDP: %s:%d -> %s:%d"%(ip_src,
udp_sport,
ip_dst,
udp_dport))
rtn_list.append({'protocol':'UDP',
'ip_src': ip_src,
'ip_dst': ip_dst,
'port_src': udp_sport,
'port_dst': udp_dport,
'dns':dns_hosts})
return rtn_list, other_listdef _add_dns(self, udpdata):
"""
Adds a DNS data flow.
@param udpdata: data inside flow
"""
dns = dpkt.dns.DNS(udpdata)
name = dns.qd[0].name
if name not in self.dns_performed:
if re.search("in-addr.arpa", name):
return False
# This is generated by time-sync of the virtual machine.
if name.strip() == "time.windows.com":
return False
entry = {}
entry["hostname"] = name
try:
socket.setdefaulttimeout(10)
ip = socket.gethostbyname(name)def resolve_over_tcp(record_type, domain, server_ip, server_port, timeout):
try:
sock = create_tcp_socket(server_ip, server_port, connect_timeout=3)
except gevent.GreenletExit:
return []
except:
if LOGGER.isEnabledFor(logging.DEBUG):
LOGGER.exception('failed to connect to %s:%s' % (server_ip, server_port))
else:
report_error('failed to connect to %s:%s' % (server_ip, server_port))
return []
try:
with contextlib.closing(sock):
sock.settimeout(timeout)
request = dpkt.dns.DNS(id=get_transaction_id(), qd=[dpkt.dns.DNS.Q(name=domain, type=record_type)])
LOGGER.debug('send request: %s' % repr(request))
data = str(request)
sock.send(struct.pack('>h', len(data)) + data)
data = sock.recv(8192)
data = data[2:]
response = dpkt.dns.DNS(data)
if response.get_rcode() & dpkt.dns.DNS_RCODE_NXDOMAIN:
raise NoSuchDomain()
if not is_right_response(server_ip, response): # filter opendns "nxdomain"
response = None
if response:
if dpkt.dns.DNS_A == record_type:
return list_ipv4_addresses(response)
elif dpkt.dns.DNS_TXT == record_type:
return [answer.text[0] for answer in response.an]
else:dpkt
fast, simple packet creation / parsing, with definitions for the basic TCP/IP protocols
