← All posts

如何写一个web服务器

目录 · Table of Contents

我们将用 Python 从零实现一个可扩展的 Web 服务器,并一步步为它加上动态页面渲染、静态文件托管、目录列表、CGI 脚本执行,以及基于责任链模式的架构重构。

完整代码可以在文末查看,也可以直接下载 web_server.py。


1. Hello, Web

先来写一个最简单的 Web 服务器。基本工作流程只有六步:

  1. 等待客户端连接服务器,并发送一个 HTTP 请求;
  2. 解析这个请求;
  3. 找出它请求的具体目标;
  4. 获取对应的数据(从磁盘读取文件,或由程序动态生成);
  5. 将数据格式化为 HTML;
  6. 把响应写回客户端。

其中第 1、2、6 步对几乎所有 HTTP 服务器来说都是一样的,Python 标准库中的 http.server 模块(在 Python 2 中叫 BaseHTTPServer)已经帮我们处理好了底层 Socket 连接与 HTTP 报文解析。我们只需要关注第 3~5 步。

下面是一个最小实现:

python
1from http.server import BaseHTTPRequestHandler, HTTPServer
2
3
4class RequestHandler(BaseHTTPRequestHandler):
5 """处理 HTTP 请求,返回一个固定的页面。"""
6
7 Page = """\
8<html>
9<body>
10<p>Hello, web!</p>
11</body>
12</html>
13"""
14
15 def do_GET(self):
16 content = self.Page.encode("utf-8")
17 self.send_response(200)
18 self.send_header("Content-Type", "text/html; charset=utf-8")
19 self.send_header("Content-Length", str(len(content)))
20 self.end_headers()
21 self.wfile.write(content)
22
23
24if __name__ == "__main__":
25 server_address = ("", 8080)
26 server = HTTPServer(server_address, RequestHandler)
27 server.serve_forever()

BaseHTTPRequestHandler 类会帮我们解析传入的 HTTP 请求,并根据请求方法调用对应的处理函数——例如收到 GET 请求时,它就会自动调用 do_GET。

我们的 RequestHandler 重写了 do_GET 方法来返回保存在类变量 Page 中的 HTML:

  • send_response(200) 发送状态码 200 OK;
  • send_header("Content-Type", ...) 告诉客户端返回的内容是 HTML;
  • send_header("Content-Length", ...) 声明响应体的字节长度(注意在 Python 3 中需要先通过 .encode("utf-8") 转为字节流 bytes);
  • end_headers() 插入一个空行,用来分隔 HTTP 响应头与响应正文;
  • self.wfile.write(content) 通过输出流把网页内容发送给客户端。

最后三行代码用来启动服务器:('', 8080) 表示在当前机器的所有网卡上监听 8080 端口;接着创建 HTTPServer 实例并调用 serve_forever() 持续循环(按 Ctrl-C 停止)。

在命令行运行它:

bash
1$ python3 server.py

用浏览器打开 http://localhost:8080,页面上会显示:

text
1Hello, web!

同时命令行终端会打印出两条访问日志:

text
1127.0.0.1 - - [24/Feb/2014 10:26:28] "GET / HTTP/1.1" 200 -
2127.0.0.1 - - [24/Feb/2014 10:26:28] "GET /favicon.ico HTTP/1.1" 200 -

第一条是因為我们没有指定具体路径,浏览器默认请求了服务器根目录 /;第二条是因为浏览器会自动请求 /favicon.ico 用来显示标签页图标。


2. 渲染动态数据

接下来我们修改代码,让服务器把当前 HTTP 请求的详细信息动态展示出来。为了保持代码整洁,我们把“生成页面内容”和“发送响应”拆分成两个方法:create_page 和 send_page:

python
1from http.server import BaseHTTPRequestHandler, HTTPServer
2
3
4class RequestHandler(BaseHTTPRequestHandler):
5
6 Page = """\
7<html>
8<body>
9<table>
10<tr> <td>Header</td> <td>Value</td> </tr>
11<tr> <td>Date and time</td> <td>{date_time}</td> </tr>
12<tr> <td>Client host</td> <td>{client_host}</td> </tr>
13<tr> <td>Client port</td> <td>{client_port}</td> </tr>
14<tr> <td>Command</td> <td>{command}</td> </tr>
15<tr> <td>Path</td> <td>{path}</td> </tr>
16</table>
17</body>
18</html>
19"""
20
21 def do_GET(self):
22 page = self.create_page()
23 self.send_page(page)
24
25 def create_page(self):
26 values = {
27 "date_time": self.date_time_string(),
28 "client_host": self.client_address[0],
29 "client_port": self.client_address[1],
30 "command": self.command,
31 "path": self.path,
32 }
33 return self.Page.format(**values)
34
35 def send_page(self, page):
36 content = page.encode("utf-8")
37 self.send_response(200)
38 self.send_header("Content-Type", "text/html; charset=utf-8")
39 self.send_header("Content-Length", str(len(content)))
40 self.end_headers()
41 self.wfile.write(content)

启动代码完全不需要改动。如果现在用浏览器访问 http://localhost:8080/something.html,会看到如下表格:

Header Value
Date and time Mon, 24 Feb 2014 17:17:12 GMT
Client host 127.0.0.1
Client port 54548
Command GET
Path /something.html

注意,此时磁盘上根本没有 something.html 这个文件,但服务器依然返回了 200 OK 而不是 404——因为 HTTP 服务器本质上只是一个接收请求并返回字节流的程序,至于返回什么完全由我们的代码决定。


3. 托管静态文件与错误处理

现在我们让服务器真正去读取磁盘上的文件并返回给客户端。

首先定义一个内部异常类 ServerException:

python
1class ServerException(Exception):
2 """用于服务器内部错误报告。"""

接着修改 do_GET 方法:

python
1def do_GET(self):
2 try:
3 # 1. 拼出请求文件在磁盘上的完整路径
4 full_path = os.getcwd() + self.path
5
6 # 2. 如果路径不存在...
7 if not os.path.exists(full_path):
8 raise ServerException(f"'{self.path}' not found")
9
10 # 3. 如果是一个普通文件...
11 elif os.path.isfile(full_path):
12 self.handle_file(full_path)
13
14 # 4. 其他暂不支持的对象(比如目录)...
15 else:
16 raise ServerException(f"Unknown object '{self.path}'")
17
18 except Exception as msg:
19 self.handle_error(msg)

这里假定服务器启动时的当前工作目录(os.getcwd())就是网站根目录,并将它与 URL 路径 self.path(总是以 / 开头)拼接起来。

  • 如果路径不存在,或者不是普通文件,就抛出 ServerException,并在 except 块中交给 handle_error 处理;
  • 如果文件存在,就调用 handle_file 读取二进制内容('rb'),再通过 send_content 发送给客户端:
python
1def handle_file(self, full_path):
2 try:
3 with open(full_path, "rb") as reader:
4 content = reader.read()
5 self.send_content(content)
6 except OSError as msg:
7 msg = f"'{self.path}' cannot be read: {msg}"
8 self.handle_error(msg)

再来编写错误处理方法 handle_error 和通用的响应发送方法 send_content。注意出错时应当返回 404 状态码,而不是默认的 200:

python
1Error_Page = """\
2<html>
3<body>
4<h1>Error accessing {path}</h1>
5<p>{msg}</p>
6</body>
7</html>
8"""
9
10def handle_error(self, msg):
11 content = self.Error_Page.format(path=self.path, msg=msg)
12 self.send_content(content, 404)
13
14def send_content(self, content, status=200):
15 if isinstance(content, str):
16 content = content.encode("utf-8")
17 self.send_response(status)
18 self.send_header("Content-Type", "text/html; charset=utf-8")
19 self.send_header("Content-Length", str(len(content)))
20 self.end_headers()
21 self.wfile.write(content)

在 send_content 中,我们兼容了 str 和 bytes 两种输入:如果是字符串(如错误页面),自动编码成 UTF-8 字节;如果是从文件直接读出的 bytes,则直接发送。


4. 目录列表与条件处理器(Case Handlers)

下一步,我们希望当 URL 指向一个目录而不是文件时,服务器也能正确处理:

  • 如果该目录下存在 index.html,就返回这个 index.html;
  • 如果没有 index.html,就自动生成一个展示该目录下所有文件的 HTML 列表页。

如果直接把这些判断全部塞进 do_GET 里,do_GET 很快就会变成一团纠缠不清的 if ... elif ... else 面条代码。更优雅的做法是退一步,把“判断当前请求属于哪种情况(test)”和“针对该情况执行什么动作(act)”抽离成独立的处理器对象(即设计模式中的责任链模式)。

重构后的 do_GET 变得非常简短:

python
1def do_GET(self):
2 try:
3 # 找出请求对应的完整路径
4 self.full_path = os.getcwd() + self.path
5
6 # 依次询问每个条件处理器能否处理当前请求
7 for case in self.Cases:
8 if case.test(self):
9 case.act(self)
10 break
11
12 except Exception as msg:
13 self.handle_error(msg)

每个条件处理器(case_*)都包含两个方法:

  • test(self, handler):检查当前请求是否命中该分支,返回布尔值;
  • act(self, handler):实际执行对应的处理动作。

一旦某个 case.test(self) 返回 True,就执行它的 act(self) 并跳出循环。我们先用三个处理器类还原上一节的行为:

python
1class case_no_file:
2 """文件或目录不存在。"""
3
4 def test(self, handler):
5 return not os.path.exists(handler.full_path)
6
7 def act(self, handler):
8 raise ServerException(f"'{handler.path}' not found")
9
10
11class case_existing_file:
12 """路径是一个已存在的文件。"""
13
14 def test(self, handler):
15 return os.path.isfile(handler.full_path)
16
17 def act(self, handler):
18 handler.handle_file(handler.full_path)
19
20
21class case_always_fail:
22 """兜底分支:以上所有情况都不匹配时报错。"""
23
24 def test(self, handler):
25 return True
26
27 def act(self, handler):
28 raise ServerException(f"Unknown object '{handler.path}'")

然后在 RequestHandler 顶部按顺序注册它们:

python
1class RequestHandler(BaseHTTPRequestHandler):
2
3 Cases = [
4 case_no_file(),
5 case_existing_file(),
6 case_always_fail(),
7 ]
8 # ...其余方法保持不变...

表面上看,代码行数变多了,还多了一层间接调用。但这种架构的威力在我们接下来添加“目录处理”时立刻就体现出来了——do_GET 一行都不用改,只需要新增两个 case 类!

第一个是目录下存在 index.html 的情况:

python
1class case_directory_index_file:
2 """当请求目录且目录下存在 index.html 时,返回 index.html。"""
3
4 def index_path(self, handler):
5 return os.path.join(handler.full_path, "index.html")
6
7 def test(self, handler):
8 return (
9 os.path.isdir(handler.full_path)
10 and os.path.isfile(self.index_path(handler))
11 )
12
13 def act(self, handler):
14 handler.handle_file(self.index_path(handler))

第二个是目录下没有 index.html 的情况——它的 test 条件只需把上面的第二项加上 not,act 则调用 handler.list_dir 来渲染目录列表:

python
1class case_directory_no_index_file:
2 """当请求目录但没有 index.html 时,返回目录文件列表。"""
3
4 def index_path(self, handler):
5 return os.path.join(handler.full_path, "index.html")
6
7 def test(self, handler):
8 return (
9 os.path.isdir(handler.full_path)
10 and not os.path.isfile(self.index_path(handler))
11 )
12
13 def act(self, handler):
14 handler.list_dir(handler.full_path)

然后在 RequestHandler 中添上 Listing_Page 模板和 list_dir 方法,并将两个新处理器放进 Cases 列表:

python
1class RequestHandler(BaseHTTPRequestHandler):
2
3 Cases = [
4 case_no_file(),
5 case_existing_file(),
6 case_directory_index_file(),
7 case_directory_no_index_file(),
8 case_always_fail(),
9 ]
10
11 Listing_Page = """\
12<html>
13<body>
14<ul>
15{0}
16</ul>
17</body>
18</html>
19"""
20
21 def list_dir(self, full_path):
22 try:
23 entries = sorted(os.listdir(full_path))
24 bullets = [f"<li>{e}</li>" for e in entries if not e.startswith(".")]
25 page = self.Listing_Page.format("\n".join(bullets))
26 self.send_content(page)
27 except OSError as msg:
28 msg = f"'{self.path}' cannot be listed: {msg}"
29 self.handle_error(msg)

5. CGI 协议:运行外部脚本

当然,大多数人不会想每次增加一个新页面都去修改 Web 服务器本身的源码。为了将“服务器”和“业务逻辑”解耦,Web 服务器很早就支持了一种名为 CGI(Common Gateway Interface,通用网关接口) 的机制:它提供了一种标准方式,让 Web 服务器可以通过运行外部程序来生成响应。

举个例子,假设我们想在网页上显示当前服务器的本地时间,完全可以写一个独立的 Python 小脚本 time.py:

python
1from datetime import datetime
2
3print(f"""\
4<html>
5<body>
6<p>Generated {datetime.now()}</p>
7</body>
8</html>""")

怎样让我们的 Web 服务器在收到 /time.py 请求时自动运行这个脚本,并把它的输出返回给浏览器呢?

得益于上一节的 Cases 设计,我们只需新增一个 case_cgi_file 处理器,判断文件路径是否以 .py 结尾:

python
1class case_cgi_file:
2 """可执行脚本文件(以 .py 结尾)。"""
3
4 def test(self, handler):
5 return (
6 os.path.isfile(handler.full_path)
7 and handler.full_path.endswith(".py")
8 )
9
10 def act(self, handler):
11 handler.run_cgi(handler.full_path)

注意要把 case_cgi_file() 放在 case_existing_file() 之前,否则 .py 文件会被当作普通静态文件直接把源码发送出去。

在 RequestHandler 中实现 run_cgi 的核心思路只有三步:

  1. 在子进程中运行该程序;
  2. 捕获子进程打印到标准输出(stdout)的内容;
  3. 将捕获到的内容发送回客户端。
python
1def run_cgi(self, full_path):
2 result = subprocess.run(
3 [sys.executable, full_path],
4 capture_output=True,
5 check=False,
6 )
7 self.send_content(result.stdout)

安全提示:让服务器直接执行 URL 指向的任意 .py 文件在实际生产环境中是非常危险的。如果攻击者能猜到服务器上某个 Python 脚本的路径(或者通过 .. 路径穿越访问上级目录),就能诱导服务器直接执行它。在文末的完整代码中,我们加入了基于 os.path.realpath 的根目录边界校验,防止通过 .. 访问网站根目录以外的文件。


6. 提炼公共基类 base_case 与架构总结

随着功能增加,代码里出现了一些重复与错位:

  • case_directory_index_file 和 case_directory_no_index_file 里都写了一遍完全相同的 index_path 方法;
  • handle_file 方法放在 RequestHandler 中,但实际上只有 case_existing_file 和 case_directory_index_file 会调用它。

解决办法很简单:为所有 case_* 处理器创建一个公共基类 base_case,把多个处理器共享的工具方法(如 handle_file 和 index_path)提炼到 base_case 中,并声明默认的 test 和 act 接口:

python
1class base_case:
2 """所有条件处理器的公共基类。"""
3
4 def handle_file(self, handler, full_path):
5 try:
6 with open(full_path, "rb") as reader:
7 content = reader.read()
8 handler.send_content(content)
9 except OSError as msg:
10 msg = f"'{handler.path}' cannot be read: {msg}"
11 handler.handle_error(msg)
12
13 def index_path(self, handler):
14 return os.path.join(handler.full_path, "index.html")
15
16 def test(self, handler):
17 raise NotImplementedError
18
19 def act(self, handler):
20 raise NotImplementedError

这样一来,每个具体的处理器子类都只剩下了最纯粹的几行代码。例如 case_existing_file:

python
1class case_existing_file(base_case):
2 """路径是一个已存在的文件。"""
3
4 def test(self, handler):
5 return os.path.isfile(handler.full_path)
6
7 def act(self, handler):
8 self.handle_file(handler, handler.full_path)

而主类 RequestHandler 也变得非常清爽:

python
1class RequestHandler(BaseHTTPRequestHandler):
2
3 Cases = [
4 case_no_file(),
5 case_cgi_file(),
6 case_existing_file(),
7 case_directory_index_file(),
8 case_directory_no_index_file(),
9 case_always_fail(),
10 ]
11
12 Error_Page = """\
13<html>
14<body>
15<h1>Error accessing {path}</h1>
16<p>{msg}</p>
17</body>
18</html>
19"""
20
21 def do_GET(self):
22 try:
23 self.full_path = os.getcwd() + self.path
24 for case in self.Cases:
25 if case.test(self):
26 case.act(self)
27 break
28 except Exception as msg:
29 self.handle_error(msg)
30
31 def handle_error(self, msg):
32 content = self.Error_Page.format(path=self.path, msg=msg)
33 self.send_content(content, 404)
34
35 def send_content(self, content, status=200):
36 if isinstance(content, str):
37 content = content.encode("utf-8")
38 self.send_response(status)
39 self.send_header("Content-Type", "text/html; charset=utf-8")
40 self.send_header("Content-Length", str(len(content)))
41 self.end_headers()
42 self.wfile.write(content)

回顾从初版代码到最终重构版的变化,可以看到两个非常实用的设计思想:

  1. 把类看作一组内聚服务的集合:RequestHandler 和 base_case 本身不做复杂的业务分支判断,而是提供基础工具(发送内容、读取文件、格式化错误页),由具体的处理器组合使用;
  2. 高可扩展性:任何人想给这个服务器扩展新功能,要么直接丢一个外部 .py CGI 脚本进去,要么新增一个几行代码的 case_* 类并加进 Cases 列表(甚至可以让服务器从配置文件动态加载 Cases),完全不必改动底层连接、HTTP 解析或文件读取的逻辑——正如标准库的 BaseHTTPRequestHandler 帮我们屏蔽了底层 Socket 细节一样。

7. 完整代码

下面是整理后的完整 web_server.py(基于 Python 3 标准库,同时补充了 MIME 类型自动识别与目录穿越防护):

展开完整 web_server.py
python
1#!/usr/bin/env python3
2"""A simple extensible HTTP web server in Python 3 (with Case Handlers and CGI)."""
3
4import mimetypes
5import os
6import subprocess
7import sys
8from http.server import BaseHTTPRequestHandler, HTTPServer
9from urllib.parse import unquote, urlsplit
10
11
12class ServerException(Exception):
13 """For internal error reporting."""
14
15
16# ----------------------------------------------------------------------
17# Case Handlers
18# ----------------------------------------------------------------------
19
20
21class base_case:
22 """Parent for case handlers."""
23
24 def handle_file(self, handler: "RequestHandler", full_path: str) -> None:
25 try:
26 with open(full_path, "rb") as reader:
27 content = reader.read()
28 content_type, _ = mimetypes.guess_type(full_path)
29 handler.send_content(
30 content,
31 status=200,
32 content_type=content_type or "text/html; charset=utf-8",
33 )
34 except OSError as msg:
35 msg_str = f"'{handler.path}' cannot be read: {msg}"
36 handler.handle_error(msg_str)
37
38 def index_path(self, handler: "RequestHandler") -> str:
39 return os.path.join(handler.full_path, "index.html")
40
41 def test(self, handler: "RequestHandler") -> bool:
42 raise NotImplementedError
43
44 def act(self, handler: "RequestHandler") -> None:
45 raise NotImplementedError
46
47
48class case_no_file(base_case):
49 """File or directory does not exist."""
50
51 def test(self, handler: "RequestHandler") -> bool:
52 return not os.path.exists(handler.full_path)
53
54 def act(self, handler: "RequestHandler") -> None:
55 raise ServerException(f"'{handler.path}' not found")
56
57
58class case_cgi_file(base_case):
59 """Something runnable (.py script)."""
60
61 def test(self, handler: "RequestHandler") -> bool:
62 return (
63 os.path.isfile(handler.full_path)
64 and handler.full_path.endswith(".py")
65 )
66
67 def act(self, handler: "RequestHandler") -> None:
68 handler.run_cgi(handler.full_path)
69
70
71class case_existing_file(base_case):
72 """File exists."""
73
74 def test(self, handler: "RequestHandler") -> bool:
75 return os.path.isfile(handler.full_path)
76
77 def act(self, handler: "RequestHandler") -> None:
78 self.handle_file(handler, handler.full_path)
79
80
81class case_directory_index_file(base_case):
82 """Serve index.html page for a directory."""
83
84 def test(self, handler: "RequestHandler") -> bool:
85 return (
86 os.path.isdir(handler.full_path)
87 and os.path.isfile(self.index_path(handler))
88 )
89
90 def act(self, handler: "RequestHandler") -> None:
91 self.handle_file(handler, self.index_path(handler))
92
93
94class case_directory_no_index_file(base_case):
95 """Serve listing for a directory without an index.html page."""
96
97 def test(self, handler: "RequestHandler") -> bool:
98 return (
99 os.path.isdir(handler.full_path)
100 and not os.path.isfile(self.index_path(handler))
101 )
102
103 def act(self, handler: "RequestHandler") -> None:
104 handler.list_dir(handler.full_path)
105
106
107class case_always_fail(base_case):
108 """Base case if nothing else worked."""
109
110 def test(self, handler: "RequestHandler") -> bool:
111 return True
112
113 def act(self, handler: "RequestHandler") -> None:
114 raise ServerException(f"Unknown object '{handler.path}'")
115
116
117# ----------------------------------------------------------------------
118# Request Handler
119# ----------------------------------------------------------------------
120
121
122class RequestHandler(BaseHTTPRequestHandler):
123 """
124 If the requested path maps to a file, that file is served.
125 If anything goes wrong, an error page is constructed.
126 """
127
128 Cases = [
129 case_no_file(),
130 case_cgi_file(),
131 case_existing_file(),
132 case_directory_index_file(),
133 case_directory_no_index_file(),
134 case_always_fail(),
135 ]
136
137 # How to display an error.
138 Error_Page = """\
139<html>
140<body>
141<h1>Error accessing {path}</h1>
142<p>{msg}</p>
143</body>
144</html>
145"""
146
147 # How to display a directory listing.
148 Listing_Page = """\
149<html>
150<body>
151<ul>
152{0}
153</ul>
154</body>
155</html>
156"""
157
158 # Classify and handle request.
159 def do_GET(self) -> None:
160 try:
161 # Parse URL path and resolve against current working directory.
162 clean_path = unquote(urlsplit(self.path).path)
163 root_dir = os.path.realpath(os.getcwd())
164 self.full_path = os.path.realpath(
165 os.path.join(root_dir, clean_path.lstrip("/"))
166 )
167
168 # Prevent directory traversal outside root_dir.
169 if os.path.commonpath([root_dir, self.full_path]) != root_dir:
170 raise ServerException(f"Access denied: '{self.path}'")
171
172 # Figure out how to handle it.
173 for case in self.Cases:
174 if case.test(self):
175 case.act(self)
176 break
177
178 # Handle errors.
179 except Exception as msg:
180 self.handle_error(msg)
181
182 def list_dir(self, full_path: str) -> None:
183 try:
184 entries = sorted(os.listdir(full_path))
185 bullets = [f"<li>{e}</li>" for e in entries if not e.startswith(".")]
186 page = self.Listing_Page.format("\n".join(bullets))
187 self.send_content(page)
188 except OSError as msg:
189 msg_str = f"'{self.path}' cannot be listed: {msg}"
190 self.handle_error(msg_str)
191
192 def run_cgi(self, full_path: str) -> None:
193 result = subprocess.run(
194 [sys.executable, full_path],
195 capture_output=True,
196 check=False,
197 )
198 if result.returncode != 0:
199 err_msg = result.stderr.decode("utf-8", errors="replace").strip()
200 self.handle_error(f"CGI script failed: {err_msg or result.returncode}")
201 return
202 self.send_content(result.stdout)
203
204 # Handle unknown objects.
205 def handle_error(self, msg: object) -> None:
206 content = self.Error_Page.format(path=self.path, msg=msg)
207 self.send_content(content, 404)
208
209 # Send actual content.
210 def send_content(
211 self,
212 content: str | bytes,
213 status: int = 200,
214 content_type: str = "text/html; charset=utf-8",
215 ) -> None:
216 if isinstance(content, str):
217 content = content.encode("utf-8")
218 self.send_response(status)
219 self.send_header("Content-Type", content_type)
220 self.send_header("Content-Length", str(len(content)))
221 self.end_headers()
222 self.wfile.write(content)
223
224
225# ----------------------------------------------------------------------
226
227if __name__ == "__main__":
228 port = int(sys.argv[1]) if len(sys.argv) > 1 else 8080
229 server_address = ("", port)
230 server = HTTPServer(server_address, RequestHandler)
231 print(f"Serving HTTP on http://localhost:{port} ...")
232 try:
233 server.serve_forever()
234 except KeyboardInterrupt:
235 print("\nServer stopped.")
236 finally:
237 server.server_close()