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
128
129
130
131
132
133
134
135
136
137
138
139
|
import socketserver
import ipaddr
import re
from lib.config import Config
class Daemon():
def __init__(self, data):
self.data = data
self.domain_regexp = re.compile("(?!-)[A-Z\d-]{1,63}(?<!-)$", re.IGNORECASE)
self.footer = None
self.header = None
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"
def get_footer(self):
if not self.footer:
f = open(Config().parser['Printing']['footer'])
self.footer = f.read()
f.close()
return self.footer
def get_header(self):
if not self.header:
f = open(Config().parser['Printing']['header'])
self.header = f.read()
f.close()
return self.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()
response += self.daemon.query(data)
response += self.daemon.get_footer()
self.request.sendall(bytes(response, 'utf-8'))
class ClassicServer(socketserver.ThreadingTCPServer):
def __init__(self, daemon):
host = Config().parser['Servers']['classic_host']
port = int(Config().parser['Servers']['classic_port'])
self.daemon = daemon
socketserver.ThreadingTCPServer.__init__(self, (host, port), WhoisHandler)
class WebServer(socketserver.ThreadingTCPServer):
def __init__(self, daemon):
self.host = Config().parser['Servers']['web_host']
self.port = int(Config().parser['Servers']['web_port'])
self.daemon = daemon
socketserver.ThreadingTCPServer.__init__(self, (self.host, self.port), WhoisHandler)
|
