使用 Vsock 与 Libzmq

# 使用 libzmq 实现 VSOCK 通信 来源：https://blog.remijouan.net/posts/libzmq-vsock-pyzmq/ ## AF_VSOCK VSOCK 并非新技术，自 Linux 4.8 内核起便已存在。对于不了解此技术的用户（https://www.youtube.com/watch?v=QzoBB9NO9Dc），其前身是 VMware 开发的 VMCI。AF_VSOCK 是一种套接字地址族，类似于 AF_INET 或 AF_UNIX。该地址族专为虚拟机（客户机）与底层虚拟机管理程序（主机）之间的通信而设计，也可用于构建同一主机上多个客户机之间的通信。过去，此类通信需通过串口实现，典型应用案例可见 https://pve.proxmox.com/wiki/Qemu-guest-agent。AF_VSOCK 的行为模式与 Unix 套接字类似，兼具 TCP/UDP 特性：支持地址与端口配置，同时提供流式套接字与数据报套接字两种通信模式。 详见手册页 https://man7.org/linux/man-pages/man7/vsock.7.html 地址采用 32 位“上下文标识符”（Context Identifier），包含以下保留 CID：VMADDR_CID_ANY、VMADDR_CID_HYPERVISOR、VMADDR_CID_LOCAL（5.6 内核新增）、VMADDR_CID_HOST。端口号同样为 32 位，1024 以下端口需要 root 权限。与 TCP/UDP 类似，可通过不同端口在同一 CID 上建立多条通信链路。 VSOCK 已存在一段时间，但支持生态逐步完善。Python/C/Golang/Rust 均提供支持，并附有基础工具及 SDK： - https://stefano-garzarella.github.io/posts/2021-01-22-socat-vsock/ - https://github.com/rust-vsock/tokio-vsock - https://mdlayher.com/blog/linux-vm-sockets-in-go/ - https://docs.python.org/3/library/socket.html#socket.AF_VSOCK - https://gitlab.com/vsock/vsock AWS 的 Nitro Enclaves 功能也采用了该技术：https://docs.aws.amazon.com/enclaves/latest/user/developing-applications-linux.html ## libzmq 中的支持情况 若计划使用 vsock，您会发现目前仅有底层绑定实现，缺乏对套接字进行抽象封装的便捷库。您可能已很久未直接操作套接字，需要重新回忆 poll、recv、send、open、listen、bind 等操作流程。幸运的是，https://zeromq.org/ 提供了优秀的解决方案：该库基于各类套接字实现了基础通信模式与最佳实践，甚至包含安全特性（http://curvezmq.org/）。 令人惊喜的是，libzmq 已有 VMCI 实现（https://libzmq.readthedocs.io/en/latest/zmq_vmci.html），但尚未支持 VSOCK。因此我直接参照 VMCI 代码提出了实现方案（https://github.com/zeromq/libzmq/pull/4822）。在 libzmq 中集成 VSOCK 将带来显著优势： - 所有支持 libzmq 绑定的语言（Python、Ruby、Node.js、Perl、Java、Lua 等）均可使用 AF_VSOCK - 可通过 libzmq 在 AF_VSOCK 上使用高级特性：Curve 认证、多种消息模式（req/rep、pub/sub 等） libzmq 因稳定性较高且迭代较慢，并不频繁发布新版本。同时作为嵌入式友好型库，支持基于特定提交的静态编译。VSOCK 功能终将纳入稳定版，目前我已 fork pyzmq 并集成最新 libzmq 提交以供使用：https://github.com/remijouannet/pyzmq-vsock。 libzmq vsock 文档：https://github.com/zeromq/libzmq/blob/master/doc/zmq_vsock.adoc ## pyzmq-vsock 示例 ### Hello World 自 Linux 5.6 起支持回环 CID（VMADDR_CID_LOCAL），无需虚拟机即可测试 VSOCK。在 ZMQ 中使用 @ 符号可直接绑定到回环地址。以下为基于 VSOCK 回环的 req/rep 套接字基础示例。 若要在真实宿主/客户机环境中测试 vsock，需在 QEMU 命令行中添加 vsock 设备（或相应 libvirt 配置 https://libvirt.org/formatdomain.html#vsock）： ```bash export CID=100 /usr/local/bin/qemu-system-x86_64 \ ... \ -device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=$CID \ ... ``` ```bash # 可能需要加载回环模块：sudo modprobe vsock_loopback python3 -m venv venv venv/bin/pip install \ https://github.com/remijouannet/pyzmq-vsock/releases/download/27.2.0.dev0%2B4649337/pyzmq-27.2.0.dev0+4649337-cp312-abi3-manylinux_2_26_x86_64.manylinux_2_28_x86_64.whl # 基于 https://zeromq.org/get-started/?language=python# 改编的 vsock 示例 cat < rep.py # # Python 实现的 Hello World 服务端 # 绑定 REP 套接字至 vsock://@:5555 # 接收客户端发送的 b"Hello"，回复 b"World" # import time import zmq context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("vsock://@:5555") while True: # 等待客户端请求 message = socket.recv() print(f"收到请求: {message}") # 执行"处理"操作 time.sleep(1) # 发送响应给客户端 socket.send(b"World") EOF cat < req.py # # Python 实现的 Hello World 客户端 # 连接 REQ 套接字至 vsock://@:5555 # 向服务端发送"Hello"，等待返回"World" # import zmq context = zmq.Context() # 与服务端通信的套接字 print("正在连接 hello world 服务端...") socket = context.socket(zmq.REQ) socket.connect("vsock://@:5555") # 发送 5 次请求，每次等待响应 for request in range(5): print(f"发送请求 {request} ...") socket.send(b"Hello") # 接收响应 message = socket.recv() print(f"收到响应 {request} [ {message} ]") EOF venv/bin/python3 rep.py & [1] 88069 venv/bin/python3 req.py 正在连接 hello world 服务端... 发送请求 0 ... 收到请求: b'Hello' 收到响应 0 [ b'World' ] 发送请求 1 ... 收到请求: b'Hello' 收到响应 1 [ b'World' ] 发送请求 2 ... 收到请求: b'Hello' 收到响应 2 [ b'World' ] 发送请求 3 ... 收到请求: b'Hello' 收到响应 3 [ b'World' ] 发送请求 4 ... 收到请求: b'Hello' 收到响应 4 [ b'World' ] ``` ### Curve 与 asyncio Hello World 示例仅为演示，实际生产环境通常需结合 asyncio 与 Curve 安全协议。以下示例改编自 pyzmq（https://github.com/zeromq/pyzmq/blob/main/examples/security/asyncio-ironhouse.py）。 **Curve 密钥生成脚本** ```bash #!/usr/bin/env python import json import zmq import zmq.auth keys_file = "keys.json" client_pub, client_priv = zmq.curve_keypair() client2_pub, client2_priv = zmq.curve_keypair() server_pub, server_priv = zmq.curve_keypair() with open(keys_file, "w") as f: json.dump( { "client": [client_pub.decode(), client_priv.decode()], "client2": [client2_pub.decode(), client2_priv.decode()], "server": [server_pub.decode(), server_priv.decode()], }, f, sort_keys=True, indent=4, ) ``` **REP 服务端** ```bash #!/usr/bin/env python import asyncio import json import logging import zmq import zmq.auth from zmq.asyncio import Context from zmq.auth.asyncio import AsyncioAuthenticator LOGGER = logging.getLogger(__name__) async def run(keys: dict) -> None: ctx = Context.instance() # 为该上下文启动认证器 auth = AsyncioAuthenticator(ctx) auth.start() auth.certs["*"] = {keys["client"][0].encode(): True} server = ctx.socket(zmq.REP) server.curve_publickey = zmq.utils.z85.decode(keys['server'][0]) server.curve_secretkey = zmq.utils.z85.decode(keys['server'][1]) server.curve_server = True # 必须在 bind 之前设置 server.bind('vsock://@:9000') msg = await server.recv() LOGGER.info(f"收到 {msg!r}") if msg == b"Hello": LOGGER.info("Ironhouse 测试通过") await server.send(b"World") # 关闭套接字 server.close() auth.stop() if __name__ == '__main__': if not zmq.has("vsock") or not zmq.has("curve"): raise RuntimeError( f"libzmq 版本 < 4.0 不支持安全特性。当前版本：{zmq.zmq_version()}" ) level = logging.DEBUG logging.basicConfig(level=level, format="[%(levelname)s] %(message)s") with open("keys.json") as f: keys = json.load(f) asyncio.run(run(keys)) ``` **REQ 客户端** ```bash #!/usr/bin/env python import asyncio import json import logging import zmq import zmq.auth from zmq.asyncio import Context from zmq.auth.asyncio import AsyncioAuthenticator LOGGER = logging.getLogger(__name__) async def run(keys: dict) -> None: ctx = Context.instance() # 为该上下文启动认证器 auth = AsyncioAuthenticator(ctx) auth.start() client = ctx.socket(zmq.REQ) client.curve_publickey = zmq.utils.z85.decode(keys['client'][0]) client.curve_secretkey = zmq.utils.z85.decode(keys['client'][1]) client.curve_serverkey = zmq.utils.z85.decode(keys['server'][0]) client.connect('vsock://@:9000') await client.send(b"Hello") reply = await client.recv() LOGGER.info(f"收到响应 {reply!r}") client.close() auth.stop() if __name__ == '__main__': if not zmq.has("vsock") or not zmq.has("curve"): raise RuntimeError( f"libzmq 版本 < 4.0 不支持安全特性。当前版本：{zmq.zmq_version()}" ) level = logging.DEBUG logging.basicConfig(level=level, format="[%(levelname)s] %(message)s") with open("keys.json") as f: keys = json.load(f) asyncio.run(run(keys)) ``` **认证成功时的输出** ```bash venv/bin/python generate_keys.py venv/bin/python rep_asyncio_curve.py [DEBUG] 使用选择器: EpollSelector [DEBUG] 启动 [DEBUG] version: b'1.0', request_id: b'1', domain: '', address: '', identity: b'', mechanism: b'CURVE' [DEBUG] ALLOWED (CURVE) domain=* client_key=b'HhdIwzo4=a}1F#eL{}rs4C1Hgx.Z4nd#/JqIasmP' [DEBUG] ZAP reply code=b'200' text=b'OK' [INFO] 收到 b'Hello' [INFO] Ironhouse 测试通过 venv/bin/python req_asyncio_curve.py [DEBUG] 使用选择器: EpollSelector [DEBUG] 启动 [INFO] 收到响应 b'World' ``` **使用未授权密钥时的输出** ```bash venv/bin/python generate_keys.py venv/bin/python rep_asyncio_curve.py [DEBUG] 使用选择器: EpollSelector [DEBUG] 启动 [DEBUG] version: b'1.0', request_id: b'1', domain: '', address: '', identity: b'', mechanism: b'CURVE' [DEBUG] DENIED (CURVE) domain=* client_key=b'P4//#^+&*nKTcb]6*u:zy ```