-
Notifications
You must be signed in to change notification settings - Fork 16
/
dnserver.py
76 lines (62 loc) · 2.3 KB
/
dnserver.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
# Copyright (c) Twisted Matrix Laboratories.
from twisted.internet import reactor, defer
from twisted.names import client, dns, error, server
import socket
# https://stackoverflow.com/questions/166506/finding-local-ip-addresses-using-pythons-stdlib
# trying three of the methods above, two of them are backup
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
s.connect(("8.8.8.8", 8022))
myip = s.getsockname()[0]
s.close()
except:
myip = socket.gethostbyname(socket.gethostname())
if myip == '127.0.0.1':
myip = socket.gethostbyname(socket.getfqdn())
class DynamicResolver(object):
"""
A resolver which calculates the answers to certain queries based on the
query type and name.
"""
def _dynamicResponseRequired(self, query):
"""
Check the query to determine if a dynamic response is required.
"""
domain_includes = ['magica-us.com', 'adjust.com', 'treasuredata.com', 'smbeat.jp', 'appmeasurements.com',
'snaa.services', 'trnox.com', 'applinzi.com', 'noahapps.jp', 'yeshen.com']
query_name = query.name.name.decode('ascii')
return any(query_name.endswith(domain) for domain in domain_includes)
def _doDynamicResponse(self, query):
"""
Calculate the response to a query.
"""
answer = dns.RRHeader(
name=query.name.name,
payload=dns.Record_A(address=myip.encode('ascii')))
answers = [answer]
authority = []
additional = []
return answers, authority, additional
def query(self, query, timeout=None):
"""
Check if the query should be answered dynamically, otherwise dispatch to
the fallback resolver.
"""
if self._dynamicResponseRequired(query):
return defer.succeed(self._doDynamicResponse(query))
else:
return defer.fail(error.DomainError())
def startDNS():
"""
Run the server.
"""
factory = server.DNSServerFactory(
clients=[DynamicResolver(), client.Resolver(resolv='resolv.conf')]
)
protocol = dns.DNSDatagramProtocol(controller=factory)
reactor.listenUDP(53, protocol)
reactor.listenTCP(53, factory)
print(f"DNS server started at {myip}")
reactor.run()
if __name__ == '__main__':
startDNS()