Reference source · 0.1.0-dev.11

interop.ts

scripts/interop.ts · Download original file

import assert from 'node:assert/strict';
import { spawnSync } from 'node:child_process';
import { fixture, PATIENT } from '../src/reference/fixtures.ts';
import { signCredential } from '../src/identity.ts';
import { open, seal } from '../src/envelope.ts';
import type { Envelope } from '../src/envelope.ts';
import { SCOPES } from '../src/authorization.ts';

function python(input: Record<string, any>): any {
  input = { ...input, status_evidence: input.status_evidence ?? Object.fromEntries([...f.trust.roots.keys()].map(issuer => [issuer, f.trust.statusEvidence(issuer)])) };
  const result = spawnSync(process.env.OPRP_PYTHON ?? '.venv/bin/python', ['python/oprp_crypto.py'], { input: JSON.stringify(input), encoding: 'utf8', maxBuffer: 5 * 1024 * 1024 });
  if (result.status !== 0) throw new Error(`Python interoperability failed: ${result.stderr || result.error?.message}`);
  return JSON.parse(result.stdout);
}
const f = fixture();
try {
  const identity = python({ mode: 'identity', subject: 'urn:oprp:demo:python-app' });
  const credential = signCredential(f.applicationAuthority, identity.public, 'application', f.clock());
  const roots = f.trust.policies;
  const initialStatus = Object.fromEntries([...f.trust.roots.keys()].map(issuer => [issuer, f.trust.statusEvidence(issuer)]));
  assert.throws(() => python({ mode: 'seal', identity, credential, roots, now: f.clock(), recipient_credential: signCredential(f.applicationAuthority, f.hostIdentity.public, 'host', f.clock()), body: { synthetic: true } }), /authority_constraint_violation/);
  assert.throws(() => python({ mode: 'seal', identity, credential, roots, status_evidence: initialStatus, now: f.started + 600, recipient_credential: f.hostCredential, body: { synthetic: true } }), /stale_issuer_status/);
  console.log('PASS: independent Python verifier rejects out-of-policy issuance and stale signed status.');
  const outgoing = seal({ operation: 'interop', synthetic: true, text: 'Bidirectional test' }, f.hostIdentity, f.hostCredential, identity.public, f.clock());
  const answer = python({ mode: 'reply', identity, credential, envelope: outgoing, roots, now: f.clock() }) as Envelope;
  const messageId = JSON.parse(Buffer.from(outgoing.protected, 'base64url').toString()).messageId;
  const opened = open(answer, f.hostIdentity, f.trust, f.clock(), identity.public.subject, messageId);
  assert.equal(opened.body.echo.text, 'Bidirectional test');
  console.log('PASS: Node/OpenSSL encrypts and signs; Python/PQClean verifies, decrypts, signs, and encrypts a response.');
  const request = python({ mode: 'seal', identity, credential, roots, now: f.clock(), recipient_credential: f.hostCredential,
    body: { operation: 'token', request: { request_id: 'python-request', patient: PATIENT, scopes: [SCOPES.demographicsRead], requested_duration_seconds: 86400, refresh_requested: true } } }) as Envelope;
  const response = f.host.handle(request, f.clock());
  const received = python({ mode: 'open', envelope: response, identity, roots, now: f.clock() });
  assert.equal(received.body.status, 'granted');
  assert.equal(received.body.authorized_until, f.clock() + 86400);
  console.log('PASS: Python-generated keys and token request work against the TypeScript Host.');
  function exchange(body: Record<string, any>): any {
    const request = python({ mode: 'seal', identity, credential, roots, now: f.clock(), recipient_credential: f.hostCredential, body }) as Envelope;
    const response = f.host.handle(request, f.clock());
    const opened = python({ mode: 'open', envelope: response, identity, roots, now: f.clock() });
    assert.equal(opened.header.inReplyTo, JSON.parse(Buffer.from(request.protected, 'base64url').toString()).messageId);
    return opened.body;
  }
  const snapshot = exchange({ operation: 'sync_snapshot', patient: PATIENT, access_token: received.body.access_token });
  assert.equal(snapshot.resources.length, 1); assert.equal(snapshot.resources[0].resource.id, PATIENT);
  assert.deepEqual(exchange({ operation: 'changes', patient: PATIENT, access_token: received.body.access_token, cursor: snapshot.cursor, limit: 1 }).events, []);
  f.advance(301);
  const refreshBody = { operation: 'refresh', refresh_token: received.body.refresh_token, request_id: 'python-rotation' };
  exchange(refreshBody); // Discard the first successful result as if its response were lost.
  const recovered = exchange(refreshBody);
  assert.equal(recovered.status, 'granted'); assert.equal(recovered.authorized_until, received.body.authorized_until);
  assert.deepEqual(exchange({ operation: 'changes', patient: PATIENT, access_token: recovered.access_token, cursor: snapshot.cursor, limit: 1 }).events, []);
  console.log('PASS: Python performs scoped snapshot/change polling and recovers a repeated refresh transaction.');
  const corrupt = { ...outgoing, ciphertext: (outgoing.ciphertext[0] === 'A' ? 'B' : 'A') + outgoing.ciphertext.slice(1) };
  assert.throws(() => python({ mode: 'open', identity, roots, status_evidence: initialStatus, now: f.started, envelope: corrupt }), /invalid_signature/);
  console.log('PASS: independent Python verifier rejects modified ciphertext.');
} finally { f.host.close(); f.trust.close(); }