Files
n_signer/documents/qubes_client_examples.md

2.5 KiB

qubes_client_examples.md

This document shows minimal caller-qube examples for invoking nsigner --listen qrexec through Qubes qrexec.

For complete protocol details (framing, JSON-RPC, error handling), see documents/CLIENT_IMPLEMENTATION.md.


1) Shell example (qrexec-client-vm + framed JSON)

This sends one get_public_key request and decodes one framed response.

#!/bin/sh
set -eu

TARGET_QUBE="nsigner-vault"
SERVICE="qubes.NsignerRpc"
REQ='{"id":"1","method":"get_public_key","params":[]}'

python3 - "$TARGET_QUBE" "$SERVICE" "$REQ" <<'PY'
import json
import struct
import subprocess
import sys

target, service, req_json = sys.argv[1], sys.argv[2], sys.argv[3]
frame = struct.pack(">I", len(req_json.encode("utf-8"))) + req_json.encode("utf-8")

p = subprocess.Popen(
    ["qrexec-client-vm", target, service],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
)
out, _ = p.communicate(frame)
if p.returncode != 0:
    raise SystemExit(f"qrexec-client-vm failed: {p.returncode}")
if len(out) < 4:
    raise SystemExit("short response (missing frame header)")

n = struct.unpack(">I", out[:4])[0]
payload = out[4:4+n]
if len(payload) != n:
    raise SystemExit("short response payload")

print(json.dumps(json.loads(payload.decode("utf-8")), indent=2))
PY

2) Python example (explicit frame helpers over qrexec stdio)

#!/usr/bin/env python3
import json
import struct
import subprocess


def frame_encode(obj: dict) -> bytes:
    payload = json.dumps(obj, separators=(",", ":")).encode("utf-8")
    return struct.pack(">I", len(payload)) + payload


def frame_decode(buf: bytes) -> dict:
    if len(buf) < 4:
        raise ValueError("missing frame header")
    n = struct.unpack(">I", buf[:4])[0]
    payload = buf[4:4 + n]
    if len(payload) != n:
        raise ValueError("short frame payload")
    return json.loads(payload.decode("utf-8"))


def call_nsigner_qrexec(target_qube: str, request: dict) -> dict:
    proc = subprocess.Popen(
        ["qrexec-client-vm", target_qube, "qubes.NsignerRpc"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    out, err = proc.communicate(frame_encode(request))
    if proc.returncode != 0:
        raise RuntimeError(f"qrexec failed ({proc.returncode}): {err.decode('utf-8', 'replace')}")
    return frame_decode(out)


if __name__ == "__main__":
    req = {"id": "1", "method": "get_public_key", "params": []}
    resp = call_nsigner_qrexec("nsigner-vault", req)
    print(json.dumps(resp, indent=2))