diff options
Diffstat (limited to 'main.py')
| -rw-r--r-- | main.py | 258 |
1 files changed, 258 insertions, 0 deletions
diff --git a/main.py b/main.py new file mode 100644 index 0000000..44c8d26 --- /dev/null +++ b/main.py @@ -0,0 +1,258 @@ +""" +The mighty silly webserver written in python for no good reason +""" + + +import time +import json +import gzip +import socket +import asyncio +import htmlmin +import aiofiles +import threading + + +# some constants +PACKET_SIZE = 2048 +PORT = 13700 # using random port cuz why not + + +# response status codes +RESPONSE = { + 200: b'OK', + 400: b'Bad Request', + 401: b'Unauthorized', + 403: b'Forbidden', + 404: b'Not Found', + 6969: b'UwU' +} + + +def get_response(code: int) -> bytes: + return str(code).encode("ascii") + RESPONSE.get(code, b':(') + + +def is_alive(sock: socket.socket) -> bool: + """ + Checks if the socket is still alive + :param sock: socket + :return: boolean (true if socket is alive, false otherwise) + """ + return getattr(sock, '_closed', False) + + +def decode_request(req: str) -> dict[str, str | list | None]: + # request dictionary + request = dict() + + # request type and path + request["type"] = req[:6].split(" ")[0] + request["path"] = req[len(request["type"]) + 1:req.find("\r\n")].split(" ")[0] + + # decode other headers + for line in req.split("\r\n")[1:]: + if len(split := line.split(":")) == 2: + key, value = split + value = value.lstrip(" ") + + # write key value pair + request[key] = value + + return request + + +class HTMLServer: + """ + The very cool webserver + """ + + def __init__(self): + self.sock: socket.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.clients: list[socket.socket] = [] + + # list of allowed paths + self.allowed_paths: dict[str, dict] = { + "/": {"path": "index.html", "encoding": "css/html"}, + "/robots.txt": {"path": "page/robots.txt", "encoding": "text"}, + "/favicon.ico": {"path": "page/favicon.ico", "encoding": "bin"}, + "/css/styles.css": {"path": None, "encoding": "css/html"}, + } + + def run(self): + """ + Function that starts the webserver + """ + + # bind the server to port and start listening + self.sock.bind(('', PORT)) + self.sock.listen() + + # start running thread + t = threading.Thread(target=self._run, daemon=True) + t.start() + + # keep alive + try: + while True: + time.sleep(0.1) + except KeyboardInterrupt: + self.sock.close() + print("Closed.") + + def _run(self): + """ + Run function for threads + :return: + """ + + asyncio.run(self.server_listener()) + + async def server_listener(self): + """ + Listens for new connections, and handles them + """ + + while True: + client, address = self.sock.accept() + self.clients.append(client) + await self.server_handle(client) + + async def server_handle(self, client: socket.socket): + """ + Handles the actual connections (clients) + :param client: connection socket + """ + + # message buffer + buffer = bytearray() + while True: + # try to fetch a message + # die otherwise + try: + message = client.recv(PACKET_SIZE) + except OSError: + break + if message == b'': + break + + # append packet to buffer + buffer += message + + # check EoF (2 blank lines) + if buffer[-4:] == b'\r\n\r\n': + # text buffer + text_buffer = buffer.decode("ascii") + + # decode request + request = decode_request(text_buffer) + + print(f"[{request['type']}] Request from client '{client.getpeername()[0]}'") + + # log that request + async with aiofiles.open("logs.log", "a") as f: + await f.write( + json.dumps( + { + "client": client.getpeername()[0], + "request": request + }, + indent=2 + ) + "\n" + ) + + # handle the request + if request["type"] == "GET": + await self.handle_get_request(client, request) + else: + await self.handle_other_request(client) + + # clear buffer + buffer.clear() + client.close() + self.clients.remove(client) + + async def handle_get_request(self, client: socket.socket, req: dict[str, str | None]): + # if it's yandex + if req.get("from") == "support@search.yandex.ru": + response = get_response(404) + data = b'Nothing...' + + # check UwU path + elif req["path"] == "/UwU" or req["path"] == "/U/w/U": + response = get_response(6969) + data = b'<h1>' + b'UwU ' * 2000 + b'</h1>' + + # otherwise check access + elif req["path"] in self.allowed_paths: + # get path + path = self.allowed_paths[req["path"]]["path"] + + # if path is None, return generic filepath + if path is None: + path = req["path"][1:] + + # check encoding + if self.allowed_paths[req["path"]]["encoding"] == "css/html": + # return text data + async with aiofiles.open(path, "r") as f: + data = htmlmin.minify(await f.read()).encode("ascii") + else: + # return binary / text data + async with aiofiles.open(path, "rb") as f: + data = await f.read() + response = get_response(200) + + # in any other case + else: + response = get_response(403) + data = b'Idk what you are trying to do here :/' + + # make headers + headers = {} + + # check if compression is supported + if req.get("Accept-Encoding"): + encoding_list = [enc.lstrip(" ") for enc in req["Accept-Encoding"].split(",")] + + # check for gzip, and add to headers if present + if "gzip" in encoding_list: + headers["Content-Encoding"] = "gzip" + + # send response + await self.send(client, response, data, headers) + client.close() + + async def handle_other_request(self, client: socket.socket): + # just say 'no' + await self.send(client, get_response(403), b'No. Don\'t do that, that\'s cringe') + client.close() + + async def send(self, client: socket.socket, response: bytes, data: bytes, headers: dict[str, str] | None = None): + # construct headers + formatted_headers = b'' + if headers is not None: + formatted_headers = "".join([f"{key}: {val}\r\n" for key, val in headers.items()]).encode("ascii") + + # check for compression + if headers.get("Content-Encoding") == "gzip": + # compress data + data = gzip.compress(data) + + # construct message + if formatted_headers == b'': + message = b'HTTP/1.1 ' + response + b'\r\n\r\n' + data + else: + message = b'HTTP/1.1 ' + response + b'\r\n' + formatted_headers + b'\r\n' + data + + # send message to client + client.sendall(message) + + +def main(): + server = HTMLServer() + server.run() + + +if __name__ == '__main__': + main() |