94 lines
2.5 KiB
Markdown
94 lines
2.5 KiB
Markdown
# 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.
|
|
|
|
```bash
|
|
#!/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)
|
|
|
|
```python
|
|
#!/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))
|
|
```
|