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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
import socketserver
import ipaddr
import re
class Daemon():
def __init__(self, data):
self.data = data
self.domain_regexp = re.compile("(?!-)[A-Z\d-]{1,63}(?<!-)$", re.IGNORECASE)
def query(self, q):
if self.is_ip(q):
result = self.search_ip(q)
elif self.is_domain(q):
result = self.search_domain(q)
else:
# Try to find a person with the given query
person = search_person(q)
if person:
result = person
else:
result = self.get_help()
return self.format_result(result)
def format_result(self, values):
"""Receive a dictionary and return a string"""
result = ""
for k, v in values.items():
if k == 'domains':
result += '\nAssociated domains\n'
for i, dom in enumerate(v):
result += ' {0}\n'.format(dom.name)
elif k == 'admins':
result += '\nAssociated administrator/s\n '
for i, adm in enumerate(v):
result += ' {0} {1} <{2}>\n'.format(adm.name, adm.surname, adm.email)
else:
result += '{0}: {1}\n'.format(k, v)
return result
def search_ip(self, ip):
result = {}
# Iterate over all IP block elements
networks = self.data.get_networks()
for key in networks:
for block in networks[key].ip_blocks:
if ipaddr.IPAddress(ip) in ipaddr.IPNetwork(block):
return networks[key].as_dict(self.data)
result['error'] = "Network not found"
return result
def search_domain(self, domain):
# Iterate over all network and check its domains
networks = self.data.get_networks()
domains = self.data.get_domains()
for network in networks.values():
if any(domains[d].name == domain for d in network.domains):
return network.as_dict(self.data)
return {'error':'Domain not found'}
# TODO
def search_person(self, query):
pass
def is_ip(self, query):
try:
ipaddr.IPAddress(query)
return True
except ValueError:
return False
def is_domain(self, hostname):
if len(hostname) > 255:
return False
if hostname[-1:] == ".":
hostname = hostname[:-1] # strip exactly one dot from the right, if present
return all(self.domain_regexp.match(x) for x in hostname.split("."))
# TODO
def get_help(self):
return "This will be the help"
# TODO
def get_footer(self):
return "This will be the footer"
# TODO
def get_header(self):
return "This will be the header"
class WhoisHandler(socketserver.BaseRequestHandler):
def setup(self):
self.daemon = self.server.daemon
def handle(self):
data = str(self.request.recv(100).strip(), 'utf-8')
print('Received: {}'.format(data))
#response = self.daemon.get_header() + " \n"
response = self.daemon.query(data)
#response += self.daemon.get_footer() + "\n"
self.request.sendall(bytes(response, 'utf-8'))
class ClassicServer(socketserver.ThreadingTCPServer):
def __init__(self, config, daemon):
host = config['Servers']['classic_host']
port = int(config['Servers']['classic_port'])
self.daemon = daemon
socketserver.ThreadingTCPServer.__init__(self, (host, port), WhoisHandler)
class WebServer(socketserver.ThreadingTCPServer):
def __init__(self, config, daemon):
self.host = config['Servers']['web_host']
self.port = int(config['Servers']['web_port'])
self.daemon = daemon
socketserver.ThreadingTCPServer.__init__(self, (self.host, self.port), WhoisHandler)
|
