Verified identity

const crypto = require('crypto')

const NAMES = ['x-pria-user-email', 'x-pria-user-id', 'x-pria-institution-id', 'x-pria-server-label', 'x-pria-timestamp', 'x-pria-nonce', 'x-pria-key-id']
function verifyPriaIdentity(signingSecret, headers, { maxAgeMs = 60 * 60 * 1000, now = Date.now() } = {}) {
    const h = {}
    for (const [k, v] of Object.entries(headers)) h[k.toLowerCase()] = String(v)
    const fields = {}
    for (const n of NAMES) if (h[n] != null) fields[n] = h[n]
    for (const r of ['x-pria-user-id', 'x-pria-institution-id', 'x-pria-server-label', 'x-pria-timestamp', 'x-pria-nonce']) if (fields[r] == null) return { ok: false, reason: 'missing-field' }
    const canonical = Object.keys(fields).sort().map((k) => `${k}=${fields[k]}`).join('\n')
    // Same order as Pria's own verifyIdentity: staleness first, then the signature — so the reason codes match.
    const ts = Number(fields['x-pria-timestamp'])
    if (!Number.isFinite(ts) || Math.abs(now - ts) > maxAgeMs) return { ok: false, reason: 'stale' }
    const expected = Buffer.from(crypto.createHmac('sha256', String(signingSecret)).update(canonical, 'utf8').digest('hex'))
    const got = Buffer.from(String(h['x-pria-signature'] || '').toLowerCase())
    // Compare byte lengths, not string lengths: timingSafeEqual throws on unequal buffers.
    if (expected.length !== got.length || !crypto.timingSafeEqual(expected, got)) return { ok: false, reason: 'bad-signature' }
    return { ok: true, keyId: fields['x-pria-key-id'] }
}

// In an Express handler (Node already lowercases req.headers):
// const result = verifyPriaIdentity(process.env.PRIA_MCP_SIGNING_SECRET, req.headers)
// if (!result.ok) return res.status(401).json({ error: result.reason })

import hashlib
import hmac
import time

NAMES = ['x-pria-user-email', 'x-pria-user-id', 'x-pria-institution-id', 'x-pria-server-label',
         'x-pria-timestamp', 'x-pria-nonce', 'x-pria-key-id']
REQUIRED = ['x-pria-user-id', 'x-pria-institution-id', 'x-pria-server-label', 'x-pria-timestamp', 'x-pria-nonce']


def verify_pria_identity(signing_secret, headers, max_age_ms=60 * 60 * 1000, now_ms=None):
    h = {str(k).lower(): str(v) for k, v in headers.items()}
    fields = {n: h[n] for n in NAMES if n in h}
    if any(r not in fields for r in REQUIRED):
        return {'ok': False, 'reason': 'missing-field'}
    canonical = '\n'.join(f'{k}={fields[k]}' for k in sorted(fields))
    # Same order as Pria's own verifier: staleness first, then the signature.
    now_ms = int(time.time() * 1000) if now_ms is None else now_ms
    try:
        ts = float(fields['x-pria-timestamp'])
    except ValueError:
        return {'ok': False, 'reason': 'stale'}
    if ts != ts or abs(now_ms - ts) > max_age_ms:  # ts != ts rejects NaN
        return {'ok': False, 'reason': 'stale'}
    expected = hmac.new(str(signing_secret).encode('utf-8'), canonical.encode('utf-8'), hashlib.sha256).hexdigest()
    got = h.get('x-pria-signature', '').lower()
    # Compare bytes: compare_digest raises TypeError on a non-ASCII str.
    if not hmac.compare_digest(expected.encode('utf-8'), got.encode('utf-8')):
        return {'ok': False, 'reason': 'bad-signature'}
    return {'ok': True, 'keyId': fields.get('x-pria-key-id')}