用 Python 构建一个 Gemini 服务器

# 我们起飞了 🚀：用 Python 构建一个 Gemini 服务器 来源：https://www.willmunn.xyz/gemini/python/2026/10/16/we-have-lift-off.html 在 Gemini 协议（https://geminiprotocol.net/）诞生 7 年之后，我发现了它。它是一种"回归第一性原理"的 Web 替代方案，旨在为独立内容创作者提供价值，远离主流网络的噪音、机器人和跟踪器。想象一下，如果 HTTP 没有 cookie，甚至除了 GET 之外没有任何请求头和方法，Web 会是什么样子。如果 HTML 不能添加脚本、样式或获取其他资源（没有 `script`、`style` 或 `img` 标签），除了标题和列表之外不存在任何文本装饰。如果 TLS 不需要受信任的第三方证书颁发机构（CA）来批准每一方。 这种极简主义对许多应用程序来说可能听起来非常基础甚至毫无用处，但其结果是一个专注于创作和分享内容的写作者和开发者社区，背后没有商业目的，没有网红，没有机器人，没有电子商务，没有订阅。在 Gemini 中，一个网站被称为 capsule（舱体），整个网络被称为 geminispace（Gemini 空间）。标准文档格式是一种精简版的 markdown 格式，称为 gemtext（https://geminiprotocol.net/docs/gemtext-specification.gmi）（Gemini 的 HTML 等价物）。 要浏览 geminispace，你需要一个 Gemini 客户端（相当于浏览器）。在尝试了几个选项之后，比如 deedum（https://github.com/snoe/deedum），我找到了 Lagrange（https://gmi.skyjake.fi/lagrange/）客户端，这是一款由在 Nokia 工作的 Jaakko Keränen（@skyjake）编写的出色软件。该客户端支持所有平台（移动端和桌面端），使用体验非常好。Gemini 上没有大公司。Lagrange 附带一个书签，指向一个"[入门指南](gemini://skyjake.fi/lagrange/getting_started.gmi)"页面，列出了一些搜索引擎，比如 keneddy，据我了解它完全运行在一个人的 Mac Mini 上，包括爬虫。 在探索了一段时间的 geminispace 之后，我想尝试实现自己的服务器。虽然 Python 可能不是实现服务器的理想语言，但我正在做越来越多的 Python 开发，所以用它来做一个有趣的项目似乎不错。本文接下来将按照我实现 Gemini 服务器的步骤进行介绍，尽量使用最少的外部依赖。 ### 监听 TCP 套接字 首先要做的就是创建一个 TCP 套接字来监听 Gemini 端口（端口号是 1965，即 Gemini 5 号火箭发射的年份）。为此我们这样做： ```python import socket as Socket port = 1965 socket = Socket.socket(Socket.AF_INET, Socket.SOCK_STREAM, 0) socket.setsockopt(Socket.SOL_SOCKET, Socket.SO_REUSEADDR, 1) socket.bind(('0.0.0.0', port)) socket.listen(5) while True: connection, client_address = socket.accept() connection.sendall(b'hello') ``` 这些都是封装了操作系统系统调用的底层 API，我们来解释一下其中的一些。 #### socket - `Socket.AF_INET` 指示我们的套接字允许来自 IPv4 地址的连接。 - `Socket.SOCK_STREAM` 表示监听 TCP 协议。 我们将套接字绑定到 `0.0.0.0`，这是一个特殊地址，表示我们希望监听**所有**网络接口。你也可以使用 `127.0.0.1`，但在 Docker 容器内运行时这会导致问题。 #### setsockopt - SOL_SOCKET（https://manpages.debian.org/trixie/manpages-dev/setsockopt.2.en.html）表示我们在套接字 API 层面设置选项（也可以在 TCP 协议层面设置选项，例如）。 - SO_REUSEADDR（https://manpages.debian.org/trixie/manpages/socket.7.en.html#SO_REUSEADDR）选项告诉操作系统可以绑定套接字，除非该地址和端口上已经有活动的套接字在监听。没有这个选项，我发现如果我终止并重启程序，就无法立即重新绑定套接字。 ### 测试你的 TCP 套接字 如果你运行上面概述的代码，可以使用 TCP 客户端来测试服务器。我建议使用 `telnet` 或 `nc`，因为你很可能已经安装了它们。运行 Python 脚本，然后在另一个终端窗口中执行以下命令之一： ``` nc 0.0.0.0 1965 telnet 0.0.0.0 1965 ``` 如果一切正常，你应该能看到 hello 被打印出来。 ### 交换有效的 Gemini 请求和响应 目前，我们的套接字服务器只返回 `hello`。为了返回一个有效的 Gemini 响应（https://geminiprotocol.net/docs/protocol-specification.gmi#responses），我们需要返回类似这样的内容： ``` 20 text/gemini # Hello ``` - 20 是"成功"状态码（类似于 HTTP 中的 200）。 - `text/gemini` 是响应数据的 MIME 类型，稍后会详细介绍 gemtext。 - 下面这一行是实际的响应数据。 一个有效的 Gemini 请求始终只包含一个绝对 URL，其数据不能超过 1024 字节。我们可以从连接对象中读取 1024 字节。如果路径是 `/hello`，我们返回 Hello；如果路径是 `/goodbye`，返回 Goodbye。对于其他情况，我们可以发送 51 "Not found"。 ```python from urllib.parse import SplitResult, urlsplit data = connection.recv(1024) url = urlsplit(data.split("\r\n")[0]) if url.path == b"/hello": connection.send(b"20 text/gemini\r\n# Hello") elif url.path == b"/goodbye": connection.send(b"20 text/gemini\r\n# Goodbye") else: connection.send(b"51 Not Found\r\n") ``` 此时，如果你用 TCP 客户端重新运行测试，你应该能够发送一个 URL，并根据路径观察到不同的响应。 ### 使用 TLS 加密 到目前为止，我们通过套接字发送的是未加密的数据（类似于纯 HTTP 请求）。Gemini 强制要求使用 TLS，因此为了让 Gemini 客户端理解我们的请求，我们需要用 TLS 证书对数据进行加密。 与 Web 不同（Web 基本上强制要求服务器证书由特定的证书颁发机构（CA）签发），在 Gemini 中使用自签名证书是很常见的。Gemini 对 TLS 采用"首次使用即信任"（TOFU）策略。这与 SSH 连接类似，当你第一次连接到某个主机时，客户端会将其证书与该主机关联起来。如果后续请求中证书发生变化，你将收到 TLS 错误（或至少是警告），表示服务器的身份已改变。 让我们生成一个自签名证书来标识我们的服务器。为此我们将使用 `openssl` 工具，它很可能也已经安装在你的系统上了。 ```sh #!/bin/sh set -eu HOST="${1:-my-gemini-host.com}" DAYS="${2:-365}" openssl req -x509 -newkey rsa:2048 -nodes -days "$DAYS" \ -keyout key.pem -out certificate.pem \ -subj "/CN=$HOST" \ -addext "subjectAltName=DNS:$DNS,DNS:localhost" chmod 600 key.pem openssl x509 -in certificate.pem -noout -subject -ext subjectAltName -dates ``` 这将生成一个自签名的 `certificate.pem` 和 `key.pem` 文件，我们可以用它们向 Gemini 客户端标识我们的服务器。 使用 ssl 模块，创建一个上下文并加载上一个脚本生成的证书和密钥文件。ssl 库允许你包装一个未加密的 TLS 套接字，这样你就可以像以前一样使用它，但数据是加密的。将你当前的代码更改为使用包装后的 `tls_connection`。 ```python import ssl context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.load_cert_chain("./certificate.pem", "./key.pem") tls_connection = context.wrap_socket(connection, server_side=True) ``` 成功了 🎉，你现在应该有了一个真正的 Gemini 客户端能够理解的服务器。你应该能够将 Lagrange（或其他任何 Gemini 客户端）指向 gemini://localhost/hello，并获得一个带有 `Hello` 标题的页面。 ### 接受用户输入 到目前为止我们所做的足以提供静态页面，但总有一天你会想要接受用户输入。要指示需要输入，服务器需要返回状态码 10，告诉客户端提示用户输入。 这比你可能习惯的 HTML 表单和 HTTP POST 要基础得多。你只有一个提示框，用户填写后，输入内容将作为整个查询字符串发送到同一个 URI。你不能发送多个查询参数或提供不同的输入类型，用户只会得到一个只有一个字段的文本框，仅此而已。 让我们添加另一个路由来回显用户输入的内容： ```python elif url.path == b"/echo": query = url.query if query == b"": tls_connection.send(b"10 What should I echo ?\r\n") else: tls_connection.send(b"20 text/gemini\r\n" + query) ``` ### 识别用户 我们现在可以接受用户输入了，但我们怎么知道输入来自哪个用户？在 HTTP 世界中有很多方法可以做到这一点，basic auth、oauth2 是常见的，但另一种方式是双向 TLS（mTLS）。 在这种场景下，服务器和客户端各自提供自己的 TLS 证书。服务器可以根据客户端的证书来识别客户端。 mTLS 在企业应用中非常常用于机器对机器的通信，但在终端用户应用中并不常见，因为为普通用户生成、维护以及指导浏览器使用客户端 TLS 证书是相当复杂和繁琐的。证书会过期，服务器需要同意信任它们，等等。 与 HTTP 不同（HTTPS 是后来附加到 TLS 上的），TLS 是 Gemini 协议的原生特性。有一组特殊的状态码（6X）与客户端证书相关。如果服务器发送状态码 60，意味着客户端必须提供客户端证书才能访问此页面。因此，创建和管理客户端证书被鼓励由 Gemini 客户端（即浏览器）支持。 Lagrange 在这方面的处理非常好。当服务器返回状态 60 时，用户会被提示要么使用之前创建的证书，要么通过一个友好的界面创建新证书，该界面隐藏了 TLS 的复杂性。一旦证书创建完成，页面可以重新加载，这次使用 mTLS，客户端提供新创建的证书。 #### Gemini 对 TLS 的处理方式 Gemini 以比 HTTPS（https://geminiprotocol.net/docs/tls-tutorial.gmi）更简化的方式使用 TLS，这并非没有安全权衡和批评（https://daniel.haxx.se/blog/2023/05/28/the-gemini-protocol-seen-by-this-http-client-person/），但相比未加密的连接确实提供了显著改善的安全性。 TLS 被用作*非对称*（公钥/私钥）加密协议，在 Gemini 中常见的做法包括： - 使用*自签名*证书而不是由证书颁发机构（CA）签发的证书，采用类似 SSH 使用 *known hosts* 的首次使用即信任（TOFU）方式。 - *忽略*（至少是客户端证书的）过期日期。 这两个要点是标准库 SSL Python 模块变得困难的地方。问题在于（据我所知）ssl 模块要求服务器在处理请求之前预先配置为信任客户端的证书颁发机构（https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_verify_locations）。 在企业/公司网络中这通常是没问题的，因为使用前一般会有预先沟通来配置证书。但对于公开网站（或 capsule）来说这不切实际。使用 Python 的 ssl 模块，由未知 CA 生成的证书会在我们能够信任它之前就被拒绝。 这促使我使用了 pyOpenSSL（https://www.pyopenssl.org/en/stable/index.html）库。 #### 重构以使用 pyOpenSSL 库处理客户端证书 关键区别在于，它允许你配置 SSL 来接受客户端证书*并*覆盖证书验证，通过 `set_verify`（https://www.pyopenssl.org/en/stable/api/ssl.html#OpenSSL.SSL.Context.set_verify）方法提供你自己的实现。 通过始终返回 `True`，我们可以将证书验证和读取逻辑上移到 Gemini 协议特定的代码中。 ```python context.set_verify(ssl.VERIFY_PEER, lambda *_: True) ``` 为了恢复带有客户端证书的 TLS 会话，我们还需要设置 session id 上下文，否则第一个请求会成功而后续请求将被拒绝（https://docs.openssl.org/4.0/man3/SSL_CTX_set_session_id_context/#warnings）。 ```python context.set_session_id(b"my_gemini_server") ``` 然后在我们的"受保护路由"中，我们可以获取证书并使用其信息在现有记录中查找用户或创建新用户。 ```python peer = tls_connection.get_peer_certificate() cert = peer.to_cryptography() names = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME) common_name = names[0].value fingerprint = cert.fingerprint(hashes.SHA256()).hex() ``` 上面代码中的指纹本质上充当证书的唯一标识符。你可以用这个值创建用户并标识他们。common_name 是一个更友好的名称，客户端在创建时会提示用户输入。你可以像用户名一样使用它来显示，但不用来标识用户。 Gemini 使用状态码 60 向客户端指示此资源需要客户端证书。在我们的受保护路由中，如果从 `get_peer_certificate()` 找不到 `peer`，我们需要返回 60 状态码，告诉客户端需要提示用户创建或使用现有证书。 值得庆幸的是，迁移到 pyOpenSSL 并不太困难。这是一个完整的示例，修改为使用 pyOpenSSL 处理客户端证书，带有一个 `/hello-user` 路由，响应证书的名称和指纹。 ```python import socket as Socket from OpenSSL import SSL as ssl from urllib.parse import SplitResult, urlsplit from cryptography.x509.oid import NameOID from cryptography.hazmat.primitives import hashes port = 1965 context = ssl.Context(ssl.TLS_SERVER_METHOD) context.use_certificate_file("./certificate.pem") context.use_privatekey_file("./key.pem") context.set_verify(ssl.VERIFY_PEER, lambda *_: True) context.set_session_id(b"my_gemini_server") socket = Socket.socket(Socket.AF_INET, Socket.SOCK_STREAM, 0) socket.setsockopt(Socket.SOL_SOCKET, Socket.SO_REUSEADDR, 1) socket.bind(("0.0.0.0", port)) socket.listen(5) while True: connection, client_address = socket.accept() tls_connection = ssl.Connection(context, connection) tls_connection.set_accept_state() tls_connection.do_handshake() data = tls_connection.recv(1024) url = urlsplit(data.split(b"\r\n")[0]) if url.path == b"/hello-user": peer = tls_connection.get_peer_certificate() if peer is None: tls_connection.send(b"60 Please authenticate\r\n") else: cert = peer.to_cryptography() names = cert.subject.get_attributes_for_oid(NameOID.COMMON_NAME) common_name = names[0].value fingerprint = cert.fingerprint(hashes.SHA256()).hex() tls_connection.send( str.encode( "\r\n".join( ( "20 text/gemini", f"# Hello {common_name}", f"## fingerprint", fingerprint, ) ) ) ) else: tls_connection.send(b"51 Not Found\r\n") ``` ### 使用异步提升性能 到目前为止我们构建的是一个单线程同步服务器。这意味着服务器一次只能处理一个请求，在处理完上一个请求之前无法接受新连接。 在你的机器上测试时这完全没问题，但一旦有多个用户，你的路由处理更复杂的逻辑并与数据库或第三方服务等其他系统交互时，就会变得力不从心。 有几种方法可以解决这个问题： - 多线程 —— 每个新连接启动一个新线程来处理请求 - 多进程 —— 类似，但启动新进程（比线程更昂贵） - async/await —— 在等待操作系统响应"I/O 任务"（如打开文件或响应网络请求）时，将控制权交还给你的程序 前两种是并行方案，后者是并发方案。这篇文章解释了两者之间的细微差别……