LibzmqでVsockを使用する

Libzmq で VSOCK を使用する

AF_VSOCK

VSOCKはLinuxカーネルに4.8から搭載されており、目新しいものではありません。VM（ゲスト）と基盤となるハイパーバイザー（ホスト）間の通信、あるいは同じホスト上の複数のゲスト間通信のために設計されたソケットアドレスファミリーです。以前は、シリアルポートを使った通信が一般的でした。AF_VSOCKは、AF_INETやAF_UNIXのようなアドレスファミリーであり、TCP/UDPのような機能を持つUnixソケットに似た感覚で利用できます。アドレスとポートがあり、ストリームとデータグラムも存在します。

man7.org/linux/man-pages/man7/vsock.7.html を参照してください。

32ビットのアドレスは「コンテキスト識別子（context identifier）」と呼ばれ、予約済みのCIDとして VMADDR_ANY, VMADDR_HYPERVISOR, VMADDR_LOCAL (5.6で新設), VMADDR_HOST があります。ポート番号は32ビットで割り当て可能ですが、1024未満のポートにはrootアクセスが必要です。TCP/UDPと同様に、同じCIDでも異なるポートを使用することで、複数の通信を確立できます。

VSOCKは以前から存在しますが、サポートはゆっくりと拡大しています。Python/C/Golang/Rustなどがサポートしており、基本的なツールやSDKも存在します。

• stefano-garzarella.github.io/posts/2021-01-22-socat-vsock/
• github.com/rust-vsock/tokio-vsock
• mdlayher.com/blog/linux-vm-sockets-in-go/
• docs.python.org/3/library/socket.html#socket.AF_VSOCK
• gitlab.com/vsock/vsock
• AWSはNitro Enclaves機能でVSOCKを使用しています docs.aws.amazon.com/enclaves/latest/user/developing-applications-linux.html

libzmq におけるサポート

VSOCKを利用しようとすると、利用できるのは低レベルのバインディングのみで、ソケットを抽象化してくれる便利なライブラリがないことにすぐに気づくでしょう。ソケットを直接開いたのは久しぶりで、poll, recv, send, open, listen, bind の方法を思い出す必要があるかもしれません。

幸いなことに、そのための素晴らしいライブラリがあります zeromq.org。これは、ほぼ全ての種類のソケット上で基本的なパターンとベストプラクティスを実装しており、セキュリティ機能 curvezmq.org も備えています。

驚いたことに、VMCIの実装は存在しましたが libzmq.readthedocs.io/en/latest/zmq_vmci.html、VSOCKの実装はありませんでした。

提案は簡単でした。VMCIのコードをほぼコピー＆ペーストしただけです github.com/zeromq/libzmq/pull/4822。

libzmqにVSOCKが追加されることで、多くの利点が得られます。

• Python, Ruby, Node.js, Perl, Java, Lua など、libzmqバインディングを持つあらゆる言語でAF_VSOCKを使用できる。
• AF_VSOCK上でlibzmqの機能（Curveによる認証、異なるメッセージパターン、Req/Rep, Pub/Subなど）を使用できる。

libzmqは安定しており、あまり頻繁に変更されないため、定期的なリリースは行われません。また、組み込みソフトウェア向けのライブラリであり、特定のコミットに対して静的にビルドすることも容易です。

VSOCKは、いつか安定版リリースに含まれるでしょう。それまでの間、最新のlibzmqコミットでビルドするためにpyzmqをフォークしました github.com/remijouannet/pyzmq-vsock。

VSOCKに関するlibzmqのドキュメントは github.com/zeromq/libzmq/blob/master/doc/zmq_vsock.adoc です。

pyzmq-vsock の例

Hello world

Linux 5.6以降、ループバックCID（VMADDR_LOCAL）が追加されたため、VMを実行せずにVSOCKをテストできます。ZMQでは「@」を使用して、このループバックに直接バインドします。

以下は、VSOCKループバック上でReq/Repソケットペアを使用する基本的な例です。実際の条件でゲストとホスト間でVSOCKをテストしたい場合は、QEMUコマンドラインにVSOCKデバイスを追加する必要があります（またはlibvirtの設定 libvirt.org/formatdomain.html#vsock）。

1export CID=100
2/usr/local/bin/qemu-system-x86_64 \
3    ...
4    -device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=$CID
5    ...

 1
 2# vsock_loopback はおそらくお使いのマシンにロードされていません
 3# sudo modprobe vsock_loopback
 4
 5python3 -m venv venv
 6
 7venv/bin/pip install \
 8    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
 9
10# zeromq.org/get-started/?language=python# から適応した例
11cat <<EOF > rep.py
12#
13#   Python での Hello World サーバー
14#   REP ソケットを vsock://@:5555 にバインドします
15#   クライアントから b"Hello" を期待し、b"World" を返します
16#
17
18import time
19import zmq
20
21context = zmq.Context()
22socket = context.socket(zmq.REP)
23socket.bind("vsock://@:5555")
24
25while True:
26    #  クライアントからの次のリクエストを待ちます
27    message = socket.recv()
28    print(f"Received request: {message}")
29
30    #  「作業」を行います
31    time.sleep(1)
32
33    #  クライアントに返信を送信します
34    socket.send(b"World")
35EOF
36
37cat <<EOF > req.py
38#
39#   Python での Hello World クライアント
40#   REQ ソケットを vsock://@:5555 に接続します
41#   サーバーに "Hello" を送信し、"World" を期待します
42#
43
44import zmq
45
46context = zmq.Context()
47
48#  サーバーと通信するためのソケット
49print("Connecting to hello world server…")
50socket = context.socket(zmq.REQ)
51socket.connect("vsock://@:5555")
52
53#  5回リクエストを送信し、毎回応答を待ちます
54for request in range(5):
55    print(f"Sending request {request} …")
56    socket.send(b"Hello")
57
58    #  応答を取得します。
59    message = socket.recv()
60    print(f"Received reply {request} [ {message} ]")
61EOF
62
63venv/bin/python3 rep.py &
64[1] 88069
65
66venv/bin/python3 req.py  
67Connecting to hello world server…
68Sending request 0 …
69Received request: b'Hello'
70Received reply 0 [ b'World' ]
71Sending request 1 …
72Received request: b'Hello'
73Received reply 1 [ b'World' ]
74Sending request 2 …
75Received request: b'Hello'
76Received reply 2 [ b'World' ]
77Sending request 3 …
78Received request: b'Hello'
79Received reply 3 [ b'World' ]
80Sending request 4 …
81Received request: b'Hello'
82Received reply 4 [ b'World' ]

Curve と asyncio

Hello World は素晴らしいですが、本番環境では asyncio と Curve を使用することが多いでしょう。以下の例は、pyzmq の例 github.com/zeromq/pyzmq/blob/main/examples/security/asyncio-ironhouse.py から取られています。

Curve キーを生成するスクリプト

 1#!/usr/bin/env python
 2
 3import json
 4
 5import zmq
 6import zmq.auth
 7
 8keys_file = "keys.json"
 9
10client_pub, client_priv = zmq.curve_keypair()
11client2_pub, client2_priv = zmq.curve_keypair()
12server_pub, server_priv = zmq.curve_keypair()
13
14with open(keys_file, "w") as f:
15    json.dump(
16        {
17            "client": [client_pub.decode(), client_priv.decode()],
18            "client2": [client2_pub.decode(), client2_priv.decode()],
19            "server": [server_pub.decode(), server_priv.decode()],
20        },
21        f,
22        sort_keys=True,
23        indent=4,
24    )

REP サーバー

 1#!/usr/bin/env python
 2
 3import asyncio
 4import json
 5import logging
 6
 7import zmq
 8import zmq.auth
 9from zmq.asyncio import Context
10from zmq.auth.asyncio import AsyncioAuthenticator
11
12LOGGER = logging.getLogger(__name__)
13
14
15async def run(keys: dict) -> None:
16    ctx = Context.instance()
17
18    # このコンテキストの認証を開始します。
19    auth = AsyncioAuthenticator(ctx)
20    auth.start()
21    auth.certs["*"].update({keys["client"][0].encode(): True})
22
23    server = ctx.socket(zmq.REP)
24
25    server.curve_publickey = zmq.utils.z85.decode(keys['server'][0])
26    server.curve_secretkey = zmq.utils.z85.decode(keys['server'][1])
27    server.curve_server = True  # bind より前に指定する必要があります
28    server.bind('vsock://@:9000')
29
30    msg = await server.recv()
31    LOGGER.info(f"Received {msg!r}")
32    if msg == b"Hello":
33        LOGGER.info("Ironhouse test OK")
34    await server.send(b"World")
35
36    # ソケットを閉じます
37    server.close()
38    auth.stop()
39
40
41if __name__ == '__main__':
42    if not zmq.has("vsock") or not zmq.has("curve"):
43        raise RuntimeError(
44            f"Security is not supported in libzmq version < 4.0. libzmq version {zmq.zmq_version()}"
45        )
46
47    level = logging.DEBUG
48
49    logging.basicConfig(level=level, format="[%(levelname)s] %(message)s")
50
51    with open("keys.json") as f:
52        keys = json.load(f)
53
54    asyncio.run(run(keys))

REQ クライアント

 1#!/usr/bin/env python
 2
 3import asyncio
 4import json
 5import logging
 6
 7import zmq
 8import zmq.auth
 9from zmq.asyncio import Context
10from zmq.auth.asyncio import AsyncioAuthenticator
11
12LOGGER = logging.getLogger(__name__)
13
14
15async def run(keys: dict) -> None:
16    ctx = Context.instance()
17
18    # このコンテキストの認証を開始します。
19    auth = AsyncioAuthenticator(ctx)
20    auth.start()
21
22    client = ctx.socket(zmq.REQ)
23    client.curve_publickey = zmq.utils.z85.decode(keys['client'][0])
24    client.curve_secretkey = zmq.utils.z85.decode(keys['client'][1])
25
26    client.curve_serverkey = zmq.utils.z85.decode(keys['server'][0])
27    client.connect('vsock://@:9000')
28
29    await client.send(b"Hello")
30    reply = await client.recv()
31    LOGGER.info(f"Received reply {reply!r}")
32
33    client.close()
34    auth.stop()
35
36
37if __name__ == '__main__':
38    if not zmq.has("vsock") or not zmq.has("curve"):
39        raise RuntimeError(
40            f"Security is not supported in libzmq version < 4.0. libzmq version {zmq.zmq_version()}"
41        )
42
43    level = logging.DEBUG
44
45    logging.basicConfig(level=level, format="[%(levelname)s] %(message)s")
46
47    with open("keys.json") as f:
48        keys = json.load(f)
49
50    asyncio.run(run(keys))

認証が成功した場合の出力

 1venv/bin/python generate_keys.py
 2
 3venv/bin/python rep_asyncio_curve.py 
 4[DEBUG] Using selector: EpollSelector
 5[DEBUG] Starting
 6[DEBUG] version: b'1.0', request_id: b'1', domain: '', address: '', identity: b'', mechanism: b'CURVE'
 7[DEBUG] ALLOWED (CURVE) domain=* client_key=b'HhdIwzo4=a}1F#eL{}rs4C1Hgx.Z4nd#/JqIasmP'
 8[DEBUG] ZAP reply code=b'200' text=b'OK'
 9[INFO] Received b'Hello'
10[INFO] Ironhouse test OK
11
12venv/bin/python req_asyncio_curve.py 
13[DEBUG] Using selector: EpollSelector
14[DEBUG] Starting
15[INFO] Received reply b'World'

無効なキーを使用した際の出力

 1venv/bin/python generate_keys.py
 2
 3venv/bin/python rep_asyncio_curve.py 
 4[DEBUG] Using selector: EpollSelector
 5[DEBUG] Starting
 6[DEBUG] version: b'1.0', request_id: b'1', domain: '', address: '', identity: b'', mechanism: b'CURVE'
 7[DEBUG] DENIED (CURVE) client_key=b'P4//#^+&*nKTcb]6*u:zy<blBUAX%IaSn=PLNG-/'
 8[DEBUG] ZAP reply code=b'400' text=b'Unknown key'
 9
10venv/bin/python req_asyncio_curve_wrong_key.py
11[DEBUG] Using selector: EpollSelector
12[DEBUG] Starting
13[INFO] No reply: server rejected this client key