Voorbeelden met pyzmq-vsock

Hallo wereld

Sinds Linux 5.6 is er een loopback CID (VMADDRCIDLOCAL), waardoor je VSOCK kunt testen zonder een virtuele machine (VM) te draaien. Gebruik @ in ZMQ om direct te binden aan deze loopback.

Hieronder volgt een basisvoorbeeld van een req/rep socket-paar over de VSOCK loopback. Als je VSOCK in een echte omgeving met een gast en host wilt testen, moet je het vsock-apparaat toevoegen aan je QEMU-opdrachtregel (of je libvirt-configuratie).

QEMU configuratie:

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

Installatie en setup:

# vsock_loopback is waarschijnlijk niet geladen op je machine
# 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

De Server (rep.py):

#
#   Hello World server in Python
#   Binds REP socket to vsock://@:5555
#   Expects b"Hello" from client, replies with b"World"
#
import time
import zmq

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

while True:
    # Wacht op het volgende verzoek van de client
    message = socket.recv()
    print(f"Received request: {message}")

    # Voer wat 'werk' uit
    time.sleep(1)

    # Stuur antwoord terug naar de client
    socket.send(b"World")

De Client (req.py):

#
#   Hello World client in Python
#   Connects REQ socket to vsock://@:5555
#   Sends "Hello" to server, expects "World" back
#
import zmq

context = zmq.Context()

# Socket om met de server te communiceren
print("Connecting to hello world server…")
socket = context.socket(zmq.REQ)
socket.connect("vsock://@:5555")

# Doe 5 verzoeken, wacht elke keer op een antwoord
for request in range(5):
    print(f"Sending request {request} …")
    socket.send(b"Hello")

    # Ontvang het antwoord
    message = socket.recv()
    print(f"Received reply {request} [ {message} ]")

Uitvoering en output:

venv/bin/python3 rep.py &
[1] 88069

venv/bin/python3 req.py
Connecting to hello world server…
Sending request 0 …
Received request: b'Hello'
Received reply 0 [ b'World' ]
Sending request 1 …
Received request: b'Hello'
Received reply 1 [ b'World' ]
Sending request 2 …
Received request: b'Hello'
Received reply 2 [ b'World' ]
Sending request 3 …
Received request: b'Hello'
Received reply 3 [ b'World' ]
Sending request 4 …
Received request: b'Hello'
Received reply 4 [ b'World' ]

Curve met asyncio

Een "Hallo wereld"-voorbeeld is leuk, maar in een productieomgeving zul je waarschijnlijk asyncio en Curve (beveiliging) gebruiken. Het volgende voorbeeld is gebaseerd op pyzmq.

Script voor het genereren van Curve-sleutels:

#!/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,
    )

De REP server:

#!/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()

    # Start een authenticator voor deze context
    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  # moet vóór bind komen
    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")

    # sluit sockets
    server.close()
    auth.stop()

if __name__ == '__main__':
    if not zmq.has("vsock") or not zmq.has("curve"):
        raise RuntimeError(
            f"Security is not supported in libzmq version < 4.0. libzmq version {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))

De REQ client:

#!/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()

    # Start een authenticator voor deze context
    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"Security is not supported in libzmq version < 4.0. libzmq version {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))

Output bij succesvolle authenticatie:

venv/bin/python generate_keys.py

venv/bin/python rep_asyncio_curve.py
[DEBUG] Using selector: EpollSelector
[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

venv/bin/python req_asyncio_curve.py
[DEBUG] Using selector: EpollSelector
[DEBUG] Starting
[INFO] Received reply b'World'

Output bij gebruik van een niet-geautoriseerde sleutel:

venv/bin/python generate_keys.py

venv/bin/python rep_asyncio_curve.py
[DEBUG] Using selector: EpollSelector
[DEBUG] Starting
[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<blBUAX%IaSn=PLNG-/'
[DEBUG] ZAP reply code=b'400' text=b'Unknown key'

venv/bin/python req_asyncio_curve_wrong_key.py
[DEBUG] Using selector: EpollSelector
[DEBUG] Starting
[INFO] No reply: server rejected this client key