Samples

// Signs a patient assertion and exchanges it for a 15-minute patient token (docs/auth.md).
// `node sign-patient-jwt.mjs`, against the local mock by default. Synthetic data only.
//
// In production your server keeps the private key, publishes the public half as a JWKS at the
// URL you registered with TeraRx, and signs one assertion each time a patient signs in to your app.

import { randomUUID } from 'node:crypto';
import { SignJWT, exportJWK, generateKeyPair } from 'jose';

const baseUrl = process.env.TERARX_BASE_URL ?? 'http://localhost:4010';
const issuer = process.env.TERARX_ISSUER ?? 'https://brand.example.com';
const audience = process.env.TERARX_AUDIENCE ?? 'https://sandbox.rx.tera.health';
const patientId = process.env.TERARX_PATIENT_ID ?? 'brand-pt-48213'; // your external_patient_id
const tenantId = process.env.TERARX_TENANT_ID; // platform partners only: the patient's brand
const kid = 'demo-2026-10';

// 1. A key pair. Generate it once and keep it; rotate by publishing a new key in your JWKS.
const { publicKey, privateKey } = await generateKeyPair('ES256', { extractable: true });
const jwks = { keys: [{ ...(await exportJWK(publicKey)), kid, alg: 'ES256', use: 'sig' }] };
console.log('JWKS to publish at your registered URL:\n' + JSON.stringify(jwks, null, 2));

// 2. The assertion: the six required claims, at most 5 minutes of lifetime, no names or health data.
// A platform partner also names the brand in `tenant_id`.
const now = Math.floor(Date.now() / 1000);
const assertion = await new SignJWT(tenantId ? { tenant_id: tenantId } : {})
  .setProtectedHeader({ alg: 'ES256', kid, typ: 'JWT' })
  .setIssuer(issuer)
  .setSubject(patientId)
  .setAudience(audience)
  .setIssuedAt(now)
  .setExpirationTime(now + 300)
  .setJti(randomUUID())
  .sign(privateKey);
console.log('\nPatient assertion:\n' + assertion);

// 3. The token exchange. No client secret: the brand's app may make this call itself.
const res = await fetch(`${baseUrl}/v1/oauth/token`, {
  method: 'POST',
  headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
  body: new URLSearchParams({
    grant_type: 'urn:ietf:params:oauth:grant-type:token-exchange',
    subject_token: assertion,
    subject_token_type: 'urn:ietf:params:oauth:token-type:jwt',
  }),
});
const body = await res.json();
console.log(`\nPOST /v1/oauth/token → ${res.status}\n` + JSON.stringify(body, null, 2));
if (!res.ok) process.exit(1);

// 4. Use the patient token on /v1/patient endpoints until it expires, then sign a new assertion.
const me = await fetch(`${baseUrl}/v1/patient`, { headers: { Authorization: `Bearer ${body.access_token}` } });
console.log(`\nGET /v1/patient → ${me.status}`);
if (!me.ok) process.exit(1);

"""Signs a patient assertion and exchanges it for a 15-minute patient token (docs/auth.md).

`python sign_patient_jwt.py`, against the local mock by default. Synthetic data only.

In production your server keeps the private key, publishes the public half as a JWKS at the
URL you registered with TeraRx, and signs one assertion each time a patient signs in to your app.
"""

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid

import jwt
from cryptography.hazmat.primitives.asymmetric import ec

BASE_URL = os.environ.get("TERARX_BASE_URL", "http://localhost:4010")
ISSUER = os.environ.get("TERARX_ISSUER", "https://brand.example.com")
AUDIENCE = os.environ.get("TERARX_AUDIENCE", "https://sandbox.rx.tera.health")
PATIENT_ID = os.environ.get("TERARX_PATIENT_ID", "brand-pt-48213")  # your external_patient_id
TENANT_ID = os.environ.get("TERARX_TENANT_ID")  # platform partners only: the patient's brand
KID = "demo-2026-10"


def request(method, path, data=None, headers=None):
    req = urllib.request.Request(BASE_URL + path, data=data, headers=headers or {}, method=method)
    try:
        with urllib.request.urlopen(req) as res:
            return res.status, json.loads(res.read() or b"null")
    except urllib.error.HTTPError as err:
        return err.code, json.loads(err.read() or b"null")


def main():
    # 1. A key pair. Generate it once and keep it; rotate by publishing a new key in your JWKS.
    private_key = ec.generate_private_key(ec.SECP256R1())
    public_jwk = json.loads(jwt.algorithms.ECAlgorithm.to_jwk(private_key.public_key()))
    jwks = {"keys": [{**public_jwk, "kid": KID, "alg": "ES256", "use": "sig"}]}
    print("JWKS to publish at your registered URL:\n" + json.dumps(jwks, indent=2))

    # 2. The assertion: the six required claims, at most 5 minutes of lifetime, no names or health data.
    # A platform partner also names the brand in `tenant_id`.
    now = int(time.time())
    claims = {"iss": ISSUER, "sub": PATIENT_ID, "aud": AUDIENCE, "iat": now, "exp": now + 300, "jti": str(uuid.uuid4())}
    if TENANT_ID:
        claims["tenant_id"] = TENANT_ID
    assertion = jwt.encode(claims, private_key, algorithm="ES256", headers={"kid": KID, "typ": "JWT"})
    print("\nPatient assertion:\n" + assertion)

    # 3. The token exchange. No client secret: the brand's app may make this call itself.
    form = urllib.parse.urlencode({
        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
        "subject_token": assertion,
        "subject_token_type": "urn:ietf:params:oauth:token-type:jwt",
    }).encode()
    status, body = request("POST", "/v1/oauth/token", form, {"Content-Type": "application/x-www-form-urlencoded"})
    print(f"\nPOST /v1/oauth/token -> {status}\n" + json.dumps(body, indent=2))
    if status != 200:
        raise SystemExit(1)

    # 4. Use the patient token on /v1/patient endpoints until it expires, then sign a new assertion.
    status, _ = request("GET", "/v1/patient", headers={"Authorization": f"Bearer {body['access_token']}"})
    print(f"\nGET /v1/patient -> {status}")
    if status != 200:
        raise SystemExit(1)


if __name__ == "__main__":
    main()

// Verifies TeraRx webhook signatures (docs/webhooks.md).
// `node verify-webhook.mjs` checks the worked example from the docs.
// `node verify-webhook.mjs --serve` starts a receiver on http://localhost:4030 that verifies each delivery
// with TERARX_WEBHOOK_SECRET.

import { createHmac, timingSafeEqual } from 'node:crypto';
import { createServer } from 'node:http';
import { fileURLToPath } from 'node:url';

const TOLERANCE_SECONDS = 300;

// rawBody is the exact bytes received, never re-serialized JSON. headers use lowercase names, as Node gives them.
export function verifyTeraRx(rawBody, headers, secret, now = Date.now()) {
  const timestamp = headers['terarx-timestamp'];
  if (!/^\d+$/.test(timestamp ?? '') || Math.abs(now / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) return false;
  const expected = createHmac('sha256', secret).update(`${timestamp}.`).update(rawBody).digest();
  // During a secret rotation the header carries two signatures; either one may match.
  return (headers['terarx-signature'] ?? '').split(',').some((part) => {
    const [version, hex] = part.trim().split('=');
    const given = Buffer.from(hex ?? '', 'hex');
    return version === 'v1' && given.length === expected.length && timingSafeEqual(given, expected);
  });
}

// The worked example in docs/webhooks.md.
export const workedExample = {
  secret: 'example_webhook_secret_new',
  oldSecret: 'example_webhook_secret_old',
  timestamp: '1793023500',
  body: '{"id":"0190f5d8-4b2d-7e6f-8a21-c2d3e4f5a6b7","type":"clinical.flag_raised","created_at":"2026-10-26T14:05:00Z","tenant_id":"a0000000-0000-4000-8000-000000000002","data":{"enrollment_id":"0190f5c2-3a1b-7c4d-8e9f-a0b1c2d3e4f5","external_patient_id":"brand-pt-48213","flag_id":"0190f5d8-4b2c-7d5e-9f10-b1c2d3e4f5a6","severity":"critical","raised_risk_to":"high","ack_due_at":"2026-10-26T16:05:00Z"}}',
  signature: 'v1=5e9f492e2cee549f0d1688a9d4a6d0505ea1687e5c293eb28a01bad16ee7aa2a',
  rotatedSignature: 'v1=5e9f492e2cee549f0d1688a9d4a6d0505ea1687e5c293eb28a01bad16ee7aa2a,v1=2e8d962ded17a921cbc548b968abc024e9fcae75fd8ff8a0f2044999b0898416',
};

function checkWorkedExample() {
  const { secret, oldSecret, timestamp, body, signature, rotatedSignature } = workedExample;
  const at = Number(timestamp) * 1000;
  const cases = [
    ['valid signature', verifyTeraRx(body, { 'terarx-timestamp': timestamp, 'terarx-signature': signature }, secret, at), true],
    ['rotation: new secret', verifyTeraRx(body, { 'terarx-timestamp': timestamp, 'terarx-signature': rotatedSignature }, secret, at), true],
    ['rotation: old secret', verifyTeraRx(body, { 'terarx-timestamp': timestamp, 'terarx-signature': rotatedSignature }, oldSecret, at), true],
    ['changed body', verifyTeraRx(body.replace('critical', 'warning'), { 'terarx-timestamp': timestamp, 'terarx-signature': signature }, secret, at), false],
    ['wrong secret', verifyTeraRx(body, { 'terarx-timestamp': timestamp, 'terarx-signature': signature }, 'not_the_secret', at), false],
    ['timestamp 6 minutes old', verifyTeraRx(body, { 'terarx-timestamp': timestamp, 'terarx-signature': signature }, secret, at + 360_000), false],
  ];
  let failed = 0;
  for (const [name, got, want] of cases) {
    if (got !== want) failed++;
    console.log(`${got === want ? '✔' : '✖'} ${name}: ${got ? 'accepted' : 'rejected'}`);
  }
  if (failed) process.exit(1);
}

function serve() {
  const secret = process.env.TERARX_WEBHOOK_SECRET;
  if (!secret) throw new Error('Set TERARX_WEBHOOK_SECRET to your webhook secret.');
  const seen = new Map(); // event id → time received. Keep ids for at least 24 hours (use Redis or your database).
  createServer((req, res) => {
    const chunks = [];
    req.on('data', (chunk) => chunks.push(chunk));
    req.on('end', () => {
      const raw = Buffer.concat(chunks);
      if (!verifyTeraRx(raw, req.headers, secret)) {
        res.writeHead(401).end();
        return;
      }
      const eventId = req.headers['terarx-event-id'];
      if (seen.has(eventId)) {
        res.writeHead(200).end(); // a retry: answer 200 without processing it again
        return;
      }
      seen.set(eventId, Date.now());
      res.writeHead(200).end(); // answer fast, then process (or queue) the event
      const event = JSON.parse(raw);
      console.log(`${event.type} ${event.id}`, event.data);
    });
  }).listen(4030, () => console.log('Receiving webhooks at http://localhost:4030'));
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  if (process.argv.includes('--serve')) serve();
  else checkWorkedExample();
}

"""Verifies TeraRx webhook signatures (docs/webhooks.md).

`python verify_webhook.py` checks the worked example from the docs.
`python verify_webhook.py --serve` starts a receiver on http://localhost:4030 that verifies each delivery
with TERARX_WEBHOOK_SECRET.
"""

from __future__ import annotations

import hashlib
import hmac
import json
import os
import sys
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

TOLERANCE_SECONDS = 300


def verify_terarx(raw_body: bytes, headers, secret: str, now: float | None = None) -> bool:
    """raw_body is the exact bytes received, never re-serialized JSON. headers is any case-insensitive mapping."""
    timestamp = headers.get("TeraRx-Timestamp") or ""
    now = time.time() if now is None else now
    if not timestamp.isdigit() or abs(now - int(timestamp)) > TOLERANCE_SECONDS:
        return False
    expected = hmac.new(secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    # During a secret rotation the header carries two signatures; either one may match.
    return any(
        part.strip().startswith("v1=") and hmac.compare_digest(part.strip()[3:], expected)
        for part in (headers.get("TeraRx-Signature") or "").split(",")
    )


# The worked example in docs/webhooks.md.
SECRET = "example_webhook_secret_new"
OLD_SECRET = "example_webhook_secret_old"
TIMESTAMP = "1793023500"
BODY = b'{"id":"0190f5d8-4b2d-7e6f-8a21-c2d3e4f5a6b7","type":"clinical.flag_raised","created_at":"2026-10-26T14:05:00Z","tenant_id":"a0000000-0000-4000-8000-000000000002","data":{"enrollment_id":"0190f5c2-3a1b-7c4d-8e9f-a0b1c2d3e4f5","external_patient_id":"brand-pt-48213","flag_id":"0190f5d8-4b2c-7d5e-9f10-b1c2d3e4f5a6","severity":"critical","raised_risk_to":"high","ack_due_at":"2026-10-26T16:05:00Z"}}'
SIGNATURE = "v1=5e9f492e2cee549f0d1688a9d4a6d0505ea1687e5c293eb28a01bad16ee7aa2a"
ROTATED = SIGNATURE + ",v1=2e8d962ded17a921cbc548b968abc024e9fcae75fd8ff8a0f2044999b0898416"


def check_worked_example():
    at = int(TIMESTAMP)
    signed = {"TeraRx-Timestamp": TIMESTAMP, "TeraRx-Signature": SIGNATURE}
    rotated = {"TeraRx-Timestamp": TIMESTAMP, "TeraRx-Signature": ROTATED}
    cases = [
        ("valid signature", verify_terarx(BODY, signed, SECRET, at), True),
        ("rotation: new secret", verify_terarx(BODY, rotated, SECRET, at), True),
        ("rotation: old secret", verify_terarx(BODY, rotated, OLD_SECRET, at), True),
        ("changed body", verify_terarx(BODY.replace(b"critical", b"warning"), signed, SECRET, at), False),
        ("wrong secret", verify_terarx(BODY, signed, "not_the_secret", at), False),
        ("timestamp 6 minutes old", verify_terarx(BODY, signed, SECRET, at + 360), False),
    ]
    failed = 0
    for name, got, want in cases:
        failed += got != want
        print(f"{'ok  ' if got == want else 'FAIL'} {name}: {'accepted' if got else 'rejected'}")
    if failed:
        raise SystemExit(1)


def serve():
    secret = os.environ.get("TERARX_WEBHOOK_SECRET")
    if not secret:
        raise SystemExit("Set TERARX_WEBHOOK_SECRET to your webhook secret.")
    seen = {}  # event id -> time received. Keep ids for at least 24 hours (use Redis or your database).

    class Receiver(BaseHTTPRequestHandler):
        def do_POST(self):
            raw = self.rfile.read(int(self.headers.get("Content-Length", 0)))
            if not verify_terarx(raw, self.headers, secret):
                self.send_response(401)
                self.end_headers()
                return
            event_id = self.headers.get("TeraRx-Event-Id")
            repeat = event_id in seen
            seen.setdefault(event_id, time.time())
            self.send_response(200)  # a retry gets 200 without processing it again
            self.end_headers()
            if not repeat:  # answer fast, then process (or queue) the event
                event = json.loads(raw)
                print(event["type"], event["id"], event["data"])

    print("Receiving webhooks at http://localhost:4030")
    HTTPServer(("127.0.0.1", 4030), Receiver).serve_forever()


if __name__ == "__main__":
    serve() if "--serve" in sys.argv else check_worked_example()