-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
57 lines (48 loc) · 1.47 KB
/
Copy pathmain.py
File metadata and controls
57 lines (48 loc) · 1.47 KB
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
import logging
import threading
from dnslib import QTYPE, RR, A
from dnslib.server import DNSServer
from ipaddress import ip_address
SLD = "visibleip.com."
IPS = ["::1", "127.0.0.1"]
def start_dns_server(ip):
server = DNSServer(dns_resolver(), port=53, address=ip, tcp=False)
logging.info(f"Starting DNS server on {ip}:53")
server.start()
class dns_resolver:
def resolve(self, request, handler):
reply = request.reply()
for question in request.questions:
if question.qtype == QTYPE.A and question.qname.matchSuffix(SLD):
rr = create_a_record(str(question.qname))
if rr: reply.add_answer(rr)
return reply
def create_a_record(query_name):
octets = query_name.split(".")
try:
ip = get_ip_from_query(octets)
return RR(
rname=query_name,
rtype=QTYPE.A,
rclass=1,
ttl=0,
rdata=A(ip)
)
except ValueError as e:
logging.error(e)
return None
def get_ip_from_query(octets):
if len(octets) < 5:
raise ValueError("Invalid query format")
ip_str = ".".join(octets[:4])
ip = ip_address(ip_str)
return str(ip)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
threads = []
for ip in IPS:
thread = threading.Thread(target=start_dns_server, args=(ip,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()