from urllib.parse import parse_qs as upunquote VERSION = "PyServ Beta 0.1" class Abort404(Exception): """Signals for a 404 error, if listened for""" def __init__(self,*args): super().__init__(*args) # Unquotes query strings. def unquote(*args): return upunquote(*args) # Lazy text sanitizer. def lazySan(s): return s.replace("<","<").replace("&","&") class HTTPServer: """HTTP Server base class""" def __init__(self): self.methods = dict() """Registers function to handle requests to a method""" def register(self,method,f): self.methods[method]=f """Returns whether a function is set to handle the given method""" def hasMethod(self,method): return method in self.methods.keys() """Returns a list of supported methods.""" def getMethods(self): ret = list(self.methods.keys()) ret.sort() return ret """Formats a response.""" def formatResponse(self,respcode,headers,content): lines = [] lines.append("HTTP/1.1 {!s} {}".format(respcode,DEFAULT_MSGS[respcode])) for h in headers: lines.append("{}: {}".format(h,headers[h])) lines.append("") # of course for line in content: lines.append(line) return "\n".join(lines)+"\n" # This function creates a stock error page. # TODO: Replace DEFAULT_MSGS with maintained database DEFAULT_MSGS = {100: "Continue",101: "Switching Protocols",102: "Processing",200: "Ok",201: "Created",202: "Accepted",203: "Non Authoritative Information",204: "No Content",205: "Reset Content",206: "Partial Content",207: "Multi Status",208: "Already Reported",226: "Im Used",300: "Multiple Choices",301: "Moved Permanently",302: "Found",303: "See Other",304: "Not Modified",305: "Use Proxy",307: "Temporary Redirect",308: "Permanent Redirect",400: "Bad Request",401: "Unauthorized",402: "Payment Required",403: "Forbidden",404: "Not Found",405: "Method Not Allowed",406: "Not Acceptable",407: "Proxy Authentication Required",408: "Request Timeout",409: "Conflict",410: "Gone",411: "Length Required",412: "Precondition Failed",413: "Request Entity Too Large",414: "Request Uri Too Long",415: "Unsupported Media Type",416: "Requested Range Not Satisfiable",417: "Expectation Failed",418: "I'm A Teapot",422: "Unprocessable Entity",423: "Locked",424: "Failed Dependency",426: "Upgrade Required",428: "Precondition Required",429: "Too Many Requests",431: "Request Header Fields Too Large",500: "Internal Server Error",501: "Not Implemented",502: "Bad Gateway",503: "Service Unavailable",504: "Gateway Timeout",505: "Http Version Not Supported",506: "Variant Also Negotiates",507: "Insufficient Storage",508: "Loop Detected",510: "Not Extended",511: "Network Authentication Required"} def abort(errorcode,message=None,headers=dict()): lines = [] if message is None: message = DEFAULT_MSGS.get(errorcode,"Unknown response code") lines.append("HTTP/1.1 {!s} {}".format(errorcode,message)) headers["Server"] = headers.get("Server",VERSION) if errorcode!=204 and "Content-Type" not in headers: headers["Content-Type"] = "text/html" for h in headers: lines.append("{}: {}".format(h,headers[h])) if errorcode==204: return "\n".join(lines)+"\n" # No content. Duh. lines.append("") lines.append("

{!s} {}

".format(errorcode, message)) lines.append("
") lines.append("

{}

".format(headers["Server"])) return "\n".join(lines)+"\n"