# SPDX-License-Identifier: Apache-2.0 # Copyright 2026 Anil Jangra """ A minimal RiskRouter client. Python 3.8+, standard library only; copy it into your product or use it as a reference for your own HTTP layer. Server to server only: your backend holds the key, never a browser. """ import hashlib import json import re import os import secrets import urllib.error import urllib.request import uuid _RECORD_KEY = re.compile(r"^[a-z][a-z0-9_]{0,63}\Z") def canonical_record(record, _depth=1): """RFC 8785 canonical JSON under the profile the specification fixes: keys a-z, 0-9 and _ starting with a letter; whole numbers within +/-(2^53 - 1) (money in integer cents); valid Unicode strings; at most 32 levels deep.""" if _depth == 1 and not isinstance(record, dict): raise ValueError("a record is a JSON object") if _depth > 32: raise ValueError("a record is nested at most 32 deep") v = record if v is None: return "null" if isinstance(v, bool): return "true" if v else "false" if isinstance(v, float): if v != v or v in (float("inf"), float("-inf")) or not v.is_integer(): raise ValueError("numbers must be whole; write money as integer cents") v = int(v) if isinstance(v, int): if abs(v) > 2 ** 53 - 1: raise ValueError("integers must be within +/-(2^53 - 1)") return str(v) if isinstance(v, str): if any(0xD800 <= ord(c) <= 0xDFFF for c in v): raise ValueError("strings must be valid Unicode") return json.dumps(v, ensure_ascii=False) if isinstance(v, (list, tuple)): return "[" + ",".join(canonical_record(x, _depth + 1) for x in v) + "]" if isinstance(v, dict): for k in v: if not isinstance(k, str) or not _RECORD_KEY.match(k): raise ValueError("key %r is not lower-case a-z, 0-9 and _ starting with a letter" % (k,)) return "{" + ",".join(json.dumps(k) + ":" + canonical_record(v[k], _depth + 1) for k in sorted(v)) + "}" raise ValueError("%s is not a JSON value a record may hold" % type(v).__name__) class RiskRouterError(Exception): def __init__(self, status, body): self.status = status self.body = body error = (body or {}).get("error", {}) super().__init__("%s %s: %s (%s)" % (status, error.get("code"), error.get("message"), error.get("docs"))) class RiskRouter: def __init__(self, api_key=None, base_url=None, timeout=10): self.api_key = api_key or os.environ.get("RISKROUTER_API_KEY") self.base_url = (base_url or os.environ.get("RISKROUTER_BASE", "https://api.riskrouter.eu")).rstrip("/") self.timeout = timeout def _call(self, method, path, body=None, key=True, headers=None): request = urllib.request.Request(self.base_url + path, method=method, data=None if body is None else json.dumps(body).encode("utf-8")) request.add_header("Accept", "application/json") if body is not None: request.add_header("Content-Type", "application/json") if key and self.api_key: request.add_header("Authorization", "Bearer " + self.api_key) for name, value in (headers or {}).items(): request.add_header(name, value) try: with urllib.request.urlopen(request, timeout=self.timeout) as response: return json.loads(response.read() or b"null"), dict(response.headers) except urllib.error.HTTPError as error: raise RiskRouterError(error.code, json.loads(error.read() or b"null")) def price(self, vertical, components): """Priced by the server, never by you. No key is sent: nothing is recorded.""" body, _ = self._call("POST", "/api/v1/quote", {"active_vertical": vertical, "selected_components": components}, key=False) return body def record_quote(self, vertical, components, idempotency_key=None): """Priced and recorded against your key. Retrying with the same idempotency key never records twice.""" idempotency_key = idempotency_key or str(uuid.uuid4()) body, headers = self._call("POST", "/api/v1/quote", {"active_vertical": vertical, "selected_components": components}, headers={"Idempotency-Key": idempotency_key}) # A 200 prices the quote even when the ledger refused the write: check it. body["recorded"] = bool(body.get("persistence", {}).get("persisted")) body["idempotency_key"] = idempotency_key return body @staticmethod def digest(record_bytes, salt=None): """SHA-256(salt || record). Keep the salt with the record; we only ever see the digest.""" salt = salt or secrets.token_bytes(16) return hashlib.sha256(salt + record_bytes).hexdigest(), salt def record_evidence(self, record_bytes, kind, salt=None, signer=None): """Record that a regulated decision existed, without sending it. With a signer (see signer.py, key registered first with register_signing_key) the claim carries your own signature inside the leaf, so nobody, including us, can say you did not make it, and we cannot have invented it. """ digest, salt = self.digest(record_bytes, salt) entry = {"record_digest": digest, "kind": kind} if signer is not None: entry.update(signer.sign_claim(kind, digest)) body, _ = self._call("POST", "/api/v2/evidence", entry) body["salt_hex"] = salt.hex() return body def record_structured(self, record, salt=None, signer=None): """Record a structured record, such as one from a regime pack (https://riskrouter.eu/packs). The record is canonicalised (RFC 8785, see canonical_record below) and salted here; only its digest and kind leave this machine. Keep the record and the returned salt_hex: with the proof, they make the bundle a supervisor checks. """ salt = salt or secrets.token_bytes(16) digest = hashlib.sha256(salt + canonical_record(record).encode("utf-8")).hexdigest() kind = record.get("kind") if not isinstance(kind, str): raise ValueError("a structured record carries its kind, the same tag sent to the log") entry = {"record_digest": digest, "kind": kind} if signer is not None: entry.update(signer.sign_claim(kind, digest)) body, _ = self._call("POST", "/api/v2/evidence", entry) body["salt_hex"] = salt.hex() return body @staticmethod def bundle(record, salt_hex, proof): """What you hand a supervisor: the record, its salt and the proof. They check it offline.""" return {"format": "riskrouter-record-bundle|1", "record": record, "salt_hex": salt_hex, "proof": proof} def register_signing_key(self, public_jwk): """Register the public half of your signing key, once. Idempotent for the same key.""" return self._call("POST", "/api/v2/evidence/keys", {"public_key": public_jwk})[0] def signing_key(self, key_id): """Anyone may fetch a registered signing key by its id; no API key is sent.""" return self._call("GET", "/api/v2/evidence/keys?key_id=" + key_id, key=False)[0] def evidence_proof(self, leaf_index, tree_size=None): suffix = "" if tree_size is None else "&tree_size=%d" % tree_size body, _ = self._call("GET", "/api/v2/evidence/proof?leaf_index=%d%s" % (leaf_index, suffix)) return body def evidence_head(self): return self._call("GET", "/api/v2/evidence/head", key=False)[0] def export(self): """Your entries, the chain they sit in, and the signed head: evidence that outlives us.""" return self._call("GET", "/api/v1/ledger/export")[0]