summaryrefslogtreecommitdiff
path: root/core.py
diff options
context:
space:
mode:
Diffstat (limited to 'core.py')
-rw-r--r--core.py127
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)
nihil fit ex nihilo