diff options
Diffstat (limited to 'core.py')
-rw-r--r-- | core.py | 127 |
1 files changed, 0 insertions, 127 deletions
diff --git a/core.py b/core.py deleted file mode 100644 index 300f6ff..0000000 --- a/core.py +++ /dev/null @@ -1,127 +0,0 @@ -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) |