AF_VSOCK と libzmq での実装ガイド

概要

AF_VSOCK は Linux カーネル 4.8 以降に導入された、仮想マシン（ゲスト）とハイパーバイザー（ホスト）間通信用のソケットアドレスファミリーです。

特徴

• VMware 由来: 元々は「VMCI」として開発されました。

  AF_INET

  や

  AF_UNIX

  に相当します。
• 通信対象:
  • ホスト上の仮想マシン（ゲスト）同士
  • ゲストと基盤となるハイパーバイザー（ホスト）間
• 機能: Unix ソケットの見た目ですが、TCP/UDP のようにアドレスとポートを設定でき、ストリーム型またはデータグラム型通信をサポートします。
• 代替手段: シリアルポート利用（例：QEMU Guest Agent）に代わる高速な通信手段です。

詳細は Linux Vsock man ページ を参照してください。

アドレスとポートの構成

AF_VSOCK は 32 ビットのアドレス（コンテキスト ID: CID）を使用します。

リザーブ済み CID

• VMADDR_CID_ANY

• VMADDR_CID_HYPERVISOR

• VMADDR_CID_LOCAL

  (Linux カーネル 5.6 以降)

• VMADDR_CID_HOST

ポート番号

• 32 ビットのポートを割り当て可能。
• 1024 未満: ループバック用またはプリビレッジポートとして使用する場合、root アクセスが必要です（TCP/UDP と同様）。
• 同じ CID でも異なるポートを使えば、複数の通信セッションを構築できます。

サポート状況とエコシステム

AF_VSOCK は以下の言語で公式にサポートされており、SDK が存在します。

• Python
• C
• Go
• Rust

主要なライブラリ・ドキュメント

socket.AF_VSOCK

libzmq (ZeroMQ) でのサポート

背景と対応策

libzmq は

poll

recv

send

追加された利点

• 多言語での利用: Python, Ruby, Node.js, Perl, Java など、libzmq バインドが存在する全言語で AF_VSOCK を利用可能。
• 高度な機能の活用: ZeroMQ 固有の機能（Curve 認証、REQ/REP パターンなど）を VSOCK 上でそのまま使用可能。

導入方法

正式な安定版リリース前のため、フォーク版

pyzmq

pyzmq-vsock

インストール例 (Python):

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

詳細ドキュメント: libzmq VSOCK ドキュメント (AOC)

Hello World 例：VSOCK ループバックテスト

Linux 5.6 以降では、仮想マシンを起動せずともループバック CID (

VMADDR_CID_LOCAL

@

環境構築

1. モジュールのロード

ループバックデバイスがない場合は、以下をロードしてください。

sudo modprobe vsock_loopback

2. 仮想マシンでの実装（オプション）

本番環境では

vhost-vsock-pci

vsock

QEMU 設定例:

/usr/local/bin/qemu-system-x86_64 \
    -device vhost-vsock-pci,id=vhost-vsock-pci0,guest-cid=$CID \
    ...

サーバーとクライアントの実装

1. REP ソケット（サーバー）

vsock://@:5555 にバインドし、クライアントから

b"Hello"

b"World"

cat <<EOF > rep.py
import time
import zmq

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("vsock://@:5555")

while True:
    # クライアントリクエストの受信
    message = socket.recv()
    print(f"Received request: {message}")

    # 処理シミュレーション
    time.sleep(1)

    # レスポンズ送信
    socket.send(b"World")
EOF

2. REQ ソケット（クライアント）

サーバーに接続し、5 回分の

b"Hello"

cat <<EOF > req.py
import zmq

context = zmq.Context()
socket = context.socket(zmq.REQ)
print("Connecting to hello world server…")
socket.connect("vsock://@:5555")

for request in range(5):
    print(f"Sending request {request} …")
    socket.send(b"Hello")
    
    # 返信の取得
    message = socket.recv()
    print(f"Received reply {request} [ {message} ]")
EOF

3. 実行結果の確認

サーバーをバックグラウンドで起動し、クライアントを実行します。

サーバー実行:

venv/bin/python3 rep.py &
# 出力: [1] 88069 (プロセス ID など)

クライアント実行:

venv/bin/python3 req.py 
# 出力例：
Connecting to hello world server…
Sending request 0 …
Received request: b'Hello'
Received reply 0 [ b'World' ]
... (5 回のループ完了)

Curve 認証と Asyncio の使用例

本番環境では、非同期処理（Asyncio）と Curve 認証 が必須となります。以下の例は ZeroMQ のセキュリティベストプラクティスに基づいています。

キー生成スクリプト

Curve 用の公開鍵・秘密鍵ペアを JSON ファイルに出力します。

#!/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) - 非同期 + Curve 認証

#!/usr/bin/env python
import asyncio
import json
import logging
import zmq
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 
    server.bind('vsock://@:9000')

    msg = await server.recv()
    LOGGER.info(f"Received {msg!r}")
    
    if msg == b"Hello":
        LOGGER.info("Ironhouse test OK")
    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()}")

    logging.basicConfig(level=logging.DEBUG, format="[%(levelname)s] %(message)s")
    
    with open("keys.json") as f:
        keys = json.load(f)
        
    asyncio.run(run(keys))

クライアント実装 (REQ) - 非同期 + Curve 認証

サーバーと通信する際に、自身の鍵とサーバーの公開鍵をセットします。

#!/usr/bin/env python
import asyncio
import json
import logging
import zmq
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"Received reply {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()}")

    logging.basicConfig(level=logging.DEBUG, format="[%(levelname)s] %(message)s")
    
    with open("keys.json") as f:
        keys = json.load(f)
        
    asyncio.run(run(keys))

実行結果の確認

認証成功時の出力

正しいキーを使用することで通信が成立し、

Ironhouse test OK

サーバー側ログ:

[DEBUG] Starting
[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] Received b'Hello'
[INFO] Ironhouse test OK

クライアント側ログ:

[DEBUG] Starting
[INFO] Received reply b'World'

認証失敗時の出力

不正なキー（または未認証のキー）を使用すると、サーバーが拒否し、クライアントは返信を受け取らない (

No reply

サーバー側ログ (DENIED):

[DEBUG] DENIED (CURVE) domain=* client_key=b'P4//#^+&*nKTcb]6*u:zy<blBUAX%IaSn=PLNG-/'
[DEBUG] ZAP reply code=b'400' text=b'Unknown key'

クライアント側ログ:

[INFO] No reply: server rejected this client key