import mimetypes import os import sys from abc import ABC, abstractmethod from datetime import datetime from time import mktime from typing import Dict from wsgiref.handlers import format_date_time from client.httpclient import FORMAT from httplib.exceptions import NotFound, Conflict, Forbidden from httplib.message import ServerMessage as Message root = os.path.join(os.path.dirname(sys.argv[0]), "public") status_message = { 200: "OK", 201: "Created", 202: "Accepted", 304: "Not Modified", 400: "Bad Request", 404: "Not Found", 500: "Internal Server Error", } def create(message: Message): if message.method == "GET": return GetCommand(message) elif message.method == "HEAD": return HeadCommand(message) elif message.method == "POST": return PostCommand(message) elif message.method == "PUT": return PutCommand(message) else: raise ValueError() class AbstractCommand(ABC): path: str headers: Dict[str, str] msg: Message def __init__(self, message: Message): self.msg = message pass @property @abstractmethod def command(self): pass def _get_date(self): now = datetime.now() stamp = mktime(now.timetuple()) return format_date_time(stamp) @abstractmethod def execute(self): pass def _build_message(self, status: int, content_type: str, body: bytes): message = f"HTTP/1.1 {status} {status_message[status]}\r\n" message += self._get_date() + "\r\n" content_length = len(body) message += f"Content-Length: {content_length}\r\n" if content_type: message += f"Content-Type: {content_type}" if content_type.startswith("text"): message += "; charset=UTF-8" message += "\r\n" elif content_length > 0: message += f"Content-Type: application/octet-stream" message += "\r\n" message = message.encode(FORMAT) if content_length > 0: message += body message += b"\r\n" return message def _get_path(self, check=True): norm_path = os.path.normpath(self.msg.target.path) if norm_path == "/": path = root + "/index.html" else: path = root + norm_path if check and not os.path.exists(path): raise NotFound() return path class AbstractModifyCommand(AbstractCommand, ABC): @property @abstractmethod def _file_mode(self): pass def execute(self): path = self._get_path(False) dir = os.path.dirname(path) if not os.path.exists(dir): raise Forbidden("Target directory does not exists!") if os.path.exists(dir) and not os.path.isdir(dir): raise Forbidden("Target directory is an existing file!") try: with open(path, mode=f"{self._file_mode}b") as file: file.write(self.msg.body) except IsADirectoryError: raise Forbidden("The target resource is a directory!") class HeadCommand(AbstractCommand): def execute(self): path = self._get_path() mime = mimetypes.guess_type(path)[0] return self._build_message(200, mime, b"") @property def command(self): return "HEAD" class GetCommand(AbstractCommand): @property def command(self): return "GET" def get_mimetype(self, path): mime = mimetypes.guess_type(path)[0] if mime: return mime try: file = open(path, "r", encoding="utf-8") file.readline() file.close() return "text/plain" except UnicodeDecodeError: return "application/octet-stream" def execute(self): path = self._get_path() mime = self.get_mimetype(path) file = open(path, "rb") buffer = file.read() file.close() return self._build_message(200, mime, buffer) class PostCommand(AbstractModifyCommand): @property def command(self): return "POST" @property def _file_mode(self): return "a" class PutCommand(AbstractModifyCommand): @property def command(self): return "PUT" @property def _file_mode(self): return "w"