Skip to content

Instantly share code, notes, and snippets.

@thislight
Created March 20, 2023 08:44
Show Gist options
  • Save thislight/b0aaa23f247d62bb6d78cc650732b214 to your computer and use it in GitHub Desktop.
Save thislight/b0aaa23f247d62bb6d78cc650732b214 to your computer and use it in GitHub Desktop.
一个非常简单的HTTP/1.1服务器例子
from socket import socket, AF_INET, SOCK_STREAM
def read_http_request_header_string(conn: socket):
buffer = bytearray()
while True:
buffer.extend(conn.recv(4096))
header_length = buffer.find(b"\r\n\r\n")
if header_length != -1:
return buffer[:header_length]
def read_http_request(conn: socket):
s = read_http_request_header_string(conn).decode("ascii")
lines = s.split("\r\n")
# Parse the first line
fstline = lines[0]
method, path, protocol = fstline.split(" ")
if protocol != "HTTP/1.1":
raise RuntimeError("unknown protocol", protocol)
method = method.lower()
# Parse headers
header = {}
if len(lines) > 1:
for line in lines[1:]:
key, value = line.split(": ")
header[key] = value
return method, path, header
STATUS_MESSAGES = {
200: "OK",
400: "Bad Request",
404: "Not Found",
500: "Server Error",
}
def build_http_response_header(status_code, headers):
lines = [
f"HTTP/1.1 {status_code} {STATUS_MESSAGES[status_code]}"
]
for key, value in headers:
lines.append(f"{key}: {value}")
lines.append("")
lines.append("")
return '\r\n'.join(lines).encode('ascii')
DEFAULT_PAGE_HTML = """<!doctype html>
<html>
<head>
<meta charset="utf-8" />
<title>Default Page</title>
</head>
<body>
<h1>Hello World!</h1>
</body>
</html>""".encode("utf-8")
def handle_http_request(conn: socket):
try:
method, path, headers = read_http_request(conn)
except Exception as e:
conn.send(build_http_response_header(400, []))
print(f"- - 400 {STATUS_MESSAGES[400]}")
raise e
if method == "get" and (path == "/" or path == "/index.html"):
headers = [
("Charset", "UTF-8"),
("Content-Length",str(len(DEFAULT_PAGE_HTML))),
("Connection", "close"),
]
conn.send(build_http_response_header(200, headers))
conn.send(DEFAULT_PAGE_HTML)
print(f"{method.upper()} {path} 200 {STATUS_MESSAGES[200]}")
else:
conn.send(build_http_response_header(404, []))
print(f"{method.upper()} {path} 404 {STATUS_MESSAGES[404]}")
def main():
with socket(AF_INET, SOCK_STREAM) as server_port:
server_port.bind(("127.0.0.1", 8989))
server_port.listen()
while True:
conn, addr = server_port.accept()
with conn:
conn.settimeout(8)
handle_http_request(conn)
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment