pip install rox-register
# Python >= 3.10; dependencies: cryptography, httpx

import json, os
from rox_register import RoxRegister

private_jwk = json.load(open("request-key.jwk"))   # private JWK dict (or a PEM string)
client = RoxRegister(
    api_key=os.environ["RR_API_KEY"],              # rr_live_ak_... (vault API key)
    request_signing_key=("ish-prod-2026-01", private_jwk),  # (kid, key material)
)

# 1 · Seal. The SDK hashes the bytes locally; the file never leaves
#     your process. Validated and either rejected immediately or
#     accepted with 202.
pdf = open("policy.pdf", "rb").read()
accepted = client.seal(
    "policy",
    content=pdf,                                 # or content_digest_hex="..."
    external_ref="POL-2026-88123",
    idempotency_key="POL-2026-88123-v1",         # safe retries for 48h
)
print(accepted["reference"])                     # rr_ref_01J8ZKQ2...
salt = accepted["salt_hex"]                      # STORE IT

# 2 · Exactly one terminal outcome follows asynchronously, anchored or
#     failed, via the record.anchored / record.failed webhooks, or poll:
record = client.get_record(accepted["record_id"])
# record["status"]: pending and batched are pre-anchor states, treat them
# identically; anchored and failed are the terminal outcomes.

# 3 · Anyone can verify the exact bytes, no API key needed:
check = client.verify_file(pdf, salt, reference=accepted["reference"])
assert check["match"] and check["result"] == "genuine"

from flask import Flask, request, abort
from rox_register import verify_webhook_signature

app = Flask(__name__)

@app.post("/rox/events")
def rox_events():
    raw = request.get_data()   # EXACT raw body bytes, before any parsing
    if not verify_webhook_signature(
        secret=WEBHOOK_SECRET,          # whsec_..., shown once at creation
        headers=request.headers,
        raw_body=raw,                   # constant-time compare + 300s freshness
    ):
        abort(401)

    event = request.get_json()          # {id, type, created_at, data}
    if event["type"] == "record.anchored":
        mark_proven(event["data"])
    if event["type"] == "record.failed":
        escalate(event["data"])
    if event["type"] == "verification.mismatch":
        alert_security(event["data"])
    return "", 200   # non-2xx deliveries are retried, then marked dead

from rox_register import hash_fields, compute_field_root

# Sealing side: hash the fields locally. Values and per-field salts
# NEVER reach Rox, only the salted hashes do.
field_hashes, field_salts, field_root = hash_fields([
    ("policy_no", "POL-2026-88123"),
    ("insured_name", "Aya Hassan"),
    ("coverage_egp", 250000),
])
client.seal("policy", content=pdf, field_hashes=field_hashes)
# Retain field_salts (and the values) alongside your document.

# Verifier side: recompute the root from the salted hashes and compare
# with field_root_hex from the issuer's record view or the RRP-1 proof
# bundle, fully offline.
assert compute_field_root([f["salted_hash_hex"] for f in field_hashes]) \
    == record["field_root_hex"]

from rox_register import AuthError, StateError, ValidationError, RoxRegisterError

try:
    client.seal("policy", content_digest_hex=digest)
except AuthError as e:        # RR-1001..1101: key, scope, JWS problems
    rotate_credentials_alert(e.code)
except ValidationError as e:  # RR-2xxx: fix your request
    fix_request(e.code, e.detail)
except StateError as e:       # vault paused, approval required/expired
    request_approval_or_unpause(e.code)
except RoxRegisterError as e: # anything else; e.retry_after on 429/5xx
    schedule_retry(e.retry_after)
# Every error carries the RFC 7807 problem document:
# e.code, e.status, e.title, e.detail