Two-device round trip

python3 -m venv .venv && source .venv/bin/activate
pip install cbor2 pynacl httpx websockets uuid7
export SCAIGRID_HOST="https://scaigrid.scailabs.ai"
export SCAIGRID_API_KEY="sgk_..."

import base64
from nacl.public import PrivateKey
from nacl.signing import SigningKey
import httpx, os

HOST = os.environ["SCAIGRID_HOST"]
KEY  = os.environ["SCAIGRID_API_KEY"]

def b64u(b: bytes) -> str:
    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()

def gen_device():
    kem = PrivateKey.generate()
    sig = SigningKey.generate()
    return {
        "kem_priv": kem, "kem_pub":  bytes(kem.public_key),
        "sig_priv": sig, "sig_pub":  bytes(sig.verify_key),
    }

def enrol(dev, name):
    r = httpx.post(f"{HOST}/v1/modules/scaiclip/devices",
        headers={"Authorization": f"Bearer {KEY}"},
        json={
            "kem_pub": b64u(dev["kem_pub"]),
            "sig_pub": b64u(dev["sig_pub"]),
            "kind": "desktop", "platform": "linux", "name": name,
        })
    r.raise_for_status()
    return r.json()["data"]["device_id"]

alice_a, alice_b = gen_device(), gen_device()
alice_a["id"] = enrol(alice_a, "Alice Laptop")
alice_b["id"] = enrol(alice_b, "Alice Phone")
print("laptop:", alice_a["id"])
print("phone:", alice_b["id"])

import asyncio, struct, cbor2
import websockets

TYPE_CHANNEL_LIST = 0x01
TYPE_CHANNEL_SUBSCRIBE = 0x02
TYPE_CLIP_PUBLISH = 0x10
TYPE_CLIP_NEW = 0x90
TYPE_OK = 0x80

def frame(type_, req_id, body):
    return struct.pack("<BBHII", 0x01, type_, 0, req_id, len(body)) + b"\0"*4 + body

def decode(raw):
    ver, type_, flags, req_id, body_len = struct.unpack("<BBHII", raw[:12])
    return type_, req_id, raw[16:16+body_len]

async def open_ws(dev):
    return await websockets.connect(
        f"{HOST.replace('https','wss')}/v1/modules/scaiclip/ws"
        f"?token={KEY}&device_id={dev['id']}",
        subprotocols=["scaiclip/1"],
    )

async def channel_list(ws):
    await ws.send(frame(TYPE_CHANNEL_LIST, 1, cbor2.dumps({}, canonical=True)))
    raw = await ws.recv()
    type_, _, body = decode(raw)
    assert type_ == TYPE_OK
    channels = cbor2.loads(body)[1]
    return channels[0][1]  # bytes(16)

from nacl.secret import SecretBox
from nacl.bindings import crypto_kx_client_session_keys, crypto_kx_keypair
import hashlib, secrets, time, uuid

def encrypt_and_wraps(plaintext: bytes, recipients: list[dict]):
    """Fresh content key; AEAD-encrypt; HPKE-wrap for each recipient.

    A full HPKE implementation is verbose — for brevity here we
    fake the wrap with libsodium's crypto_box.  See the ScaiClip
    reference HPKE module for the RFC 9180 flow used in production.
    """
    from nacl.public import Box, PublicKey
    content_key = secrets.token_bytes(32)
    nonce = secrets.token_bytes(24)
    ciphertext = SecretBox(content_key).encrypt(plaintext, nonce)
    wraps = []
    for r in recipients:
        wrap = Box(r["kem_priv"], PublicKey(r["kem_pub"])).encrypt(content_key)
        wraps.append({
            1: bytes.fromhex(r["id"]) if len(r["id"]) == 32 else r["id"],
            2: bytes(r["kem_pub"]),  # hpke_enc placeholder
            3: bytes(wrap),
            4: 1,  # key_version
        })
    return ciphertext, wraps, hashlib.sha256(ciphertext).digest()

def build_envelope(*, sender_dev, channel_id_bytes, plaintext, ciphertext,
                   payload_hash):
    from uuid_utils import uuid7
    env = {
        1: uuid7().bytes,
        2: channel_id_bytes,
        4: "usr_alice",
        5: bytes.fromhex(sender_dev["id"]) if len(sender_dev["id"]) == 32 else sender_dev["id"],
        6: int(time.time() * 1000),
        8: 0,   # text
        9: 0,   # inline
        10: 0,  # no flags
        11: 3600,
        12: 1,
        14: payload_hash,
    }
    # Canonical CBOR of keys 1-14 is the signature input.
    from modules.scaiclip.services.envelope_codec import build_sig_input
    sig_input = build_sig_input(env)
    env[15] = sender_dev["sig_priv"].sign(sig_input).signature
    return env

async def publish(ws, sender_dev, channel_id_bytes, plaintext,
                  recipients):
    ct, wraps, phash = encrypt_and_wraps(plaintext, recipients)
    env = build_envelope(
        sender_dev=sender_dev, channel_id_bytes=channel_id_bytes,
        plaintext=plaintext, ciphertext=ct, payload_hash=phash,
    )
    body = cbor2.dumps({1: env, 2: ct}, canonical=True)
    await ws.send(frame(TYPE_CLIP_PUBLISH, 100, body))
    _, _, resp = decode(await ws.recv())
    return cbor2.loads(resp)

async def main():
    laptop_ws = await open_ws(alice_a)
    phone_ws  = await open_ws(alice_b)

    personal_id = await channel_list(phone_ws)
    print("personal channel:", personal_id.hex())

    # Publish "hello from the phone".
    resp = await publish(
        phone_ws, alice_b, personal_id,
        b"hello from the phone",
        recipients=[alice_a, alice_b],
    )
    print("published seq:", resp[1])

    # Laptop should receive clip.new.
    raw = await asyncio.wait_for(laptop_ws.recv(), timeout=2)
    type_, _, body = decode(raw)
    assert type_ == TYPE_CLIP_NEW
    print("laptop got clip:", cbor2.loads(body))

asyncio.run(main())

laptop: b6a4c9…
phone:  1c2d8f…
personal channel: 3f9a1c…
published seq: 1
laptop got clip: {1: {…envelope…}, 2: b'\x…ciphertext…'}