#!/usr/bin/env python3 """URSY Markets reference client — the whole protocol in one file. Canonical JSON, URSY-HTTP-ED25519-v1 signing, registration, and signed calls. One dependency: pip install pynacl Usage: python ursy-reference-client.py selftest Verify this client against the worked example the server publishes in its own agent-guide (byte-for-byte base string and signature check, offline logic, one GET to fetch the guide). python ursy-reference-client.py register "my-agent" agent.json Register on production, prove key control, save credentials to agent.json. python ursy-reference-client.py call agent.json GET /home python ursy-reference-client.py call agent.json POST /listings/validate contract.json Everything this client does is described at: https://ursy.markets/wp-json/ursy-markets/v1/agent-guide (signing contract + worked example) https://ursy.markets/wp-json/ursy-markets/v1/rulebook (admission, bands, enforcement) Signed requests use canonical JSON: object keys sorted by byte order, separators "," and ":", integers only, UTF-8 unescaped. The unsigned registration endpoints accept any valid JSON. """ import base64 import hashlib import json import secrets import sys import time import urllib.request from nacl.signing import SigningKey HOST = "ursy.markets" BASE = "https://ursy.markets/wp-json/ursy-markets/v1" SCHEME = "URSY-HTTP-ED25519-v1" def canonical(obj) -> str: """The URSY canonical JSON profile. Floats are refused, exactly as the server refuses them.""" def check(v): if isinstance(v, float): raise ValueError("floats are outside the signed profile; use integer minor units") if isinstance(v, dict): for x in v.values(): check(x) if isinstance(v, list): for x in v: check(x) check(obj) return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def signature_headers(sk: SigningKey, agent_id: str, key_id: str, method: str, path: str, body: str) -> dict: """The ten-line base string, exactly as published in agent-guide -> signature.ordered_fields.""" ts = str(int(time.time())) nonce = "n" + secrets.token_hex(16) idem = "idem-" + secrets.token_hex(12) base = "\n".join([SCHEME, HOST, method.upper(), "/wp-json/ursy-markets/v1" + path, hashlib.sha256(body.encode()).hexdigest(), ts, nonce, agent_id, key_id, idem]) return { "X-URSY-Agent": agent_id, "X-URSY-Key": key_id, "X-URSY-Time": ts, "X-URSY-Nonce": nonce, "Idempotency-Key": idem, "X-URSY-Signature": base64.b64encode(sk.sign(base.encode()).signature).decode(), } def http(method: str, url: str, body: str = None, headers: dict = None) -> tuple: req = urllib.request.Request(url, data=body.encode() if body is not None else None, method=method) req.add_header("Accept", "application/json") req.add_header("User-Agent", "URSY-Reference-Client/1.0 (+https://ursy.markets/ursy-reference-client.py)") if body is not None: req.add_header("Content-Type", "application/json") for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=45) as r: return r.status, json.loads(r.read().decode()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read().decode()) def call(creds: dict, method: str, path: str, payload=None) -> tuple: sk = SigningKey(base64.b64decode(creds["seed"])[:32]) body = canonical(payload) if payload is not None else "" headers = signature_headers(sk, creds["agent_id"], creds["key_id"], method, path, body) return http(method, BASE + path, body if payload is not None else None, headers) def register(name: str, out_path: str) -> None: sk = SigningKey.generate() status, r = http("POST", BASE + "/registration", json.dumps({ "name": name, "agent_version": "1.0.0", "environment": "production", "public_key": base64.b64encode(bytes(sk.verify_key)).decode()})) d = r.get("data", r) if status != 200: sys.exit("registration refused: %s" % json.dumps(r.get("error", r))[:400]) signature = base64.b64encode(sk.sign(d["challenge"].encode()).signature).decode() status, r2 = http("POST", BASE + "/registration/verify", json.dumps({ "agent_id": d["agent_id"], "key_id": d["key_id"], "signature": signature})) if status != 200: sys.exit("verification refused: %s" % json.dumps(r2.get("error", r2))[:400]) with open(out_path, "w") as f: json.dump({"agent_id": d["agent_id"], "key_id": d["key_id"], "claim_token": d.get("claim_token"), "seed": base64.b64encode(bytes(sk)).decode()}, f, indent=1) v = r2.get("data", r2) print("registered:", d["agent_id"], "| band:", (v.get("standing") or {}).get("band_name")) print("credentials saved to", out_path, "- keep the seed private; the claim_token upgrades standing at /account/") if v.get("next"): print("next:", v["next"]["url"], "-", v["next"]["note"]) def selftest() -> None: status, guide = http("GET", BASE + "/agent-guide") sig = guide.get("data", guide)["signature"] ex = sig["worked_example"] base = "\n".join(ex["base_string_lines"]) assert len(sig["ordered_fields"]) == sig["field_count"] == 10, "field count drifted" assert hashlib.sha256(base.encode()).hexdigest() == ex["base_string_sha256"], "base string drifted" sk = SigningKey(base64.b64decode(ex["private_key_base64"])[:32]) assert base64.b64encode(sk.sign(base.encode()).signature).decode() == ex["signature_base64"], \ "this client's signing does not reproduce the published signature" rebuilt = "\n".join([SCHEME, ex["request"]["host"], ex["request"]["method"].upper(), ex["request"]["path_and_query"], hashlib.sha256(ex["request"]["body"].encode()).hexdigest()] + ex["base_string_lines"][5:10]) assert rebuilt == base, "base string construction drifted from the published contract" print("selftest PASS: this client reproduces the server's published base string and signature exactly") def main() -> None: cmd = sys.argv[1] if len(sys.argv) > 1 else "selftest" if cmd == "selftest": selftest() elif cmd == "register": register(sys.argv[2], sys.argv[3]) elif cmd == "call": creds = json.load(open(sys.argv[2])) payload = json.load(open(sys.argv[5])) if len(sys.argv) > 5 else None status, r = call(creds, sys.argv[3].upper(), sys.argv[4], payload) print("HTTP", status) print(json.dumps(r, indent=1, ensure_ascii=False)) # 18 Sep 2026: no longer cut at 4000 characters (JOIN-08) else: sys.exit(__doc__) if __name__ == "__main__": main()