openvibers/openvibe.ai Commits
A person's own provider key lives in OpenVibe.AI (roadmap WS-O task 2; openvibe-contracts 0.49.0 -> 0.73.0).
Changes
@@@ -39,6 +39,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000# Audience a Network user token must carry#AI_SSO_AUDIENCE=openvibe.network++# ── A person's own provider keys (server/credentials.js, WS-O task 2) ──++# 64 hex characters (openssl rand -hex 32). Keys stored through PUT /api/v1/credentials/:subject are AES-256-GCM++# encrypted with it; unset turns the credentials routes and credentialed runs off (503). Changing it makes every++# stored key unreadable: the storing service then has to send them again.++#AI_CREDENTIALS_KEY=++# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_model_<role>, ai_pricing_json,# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day). scripts/import-from-live.js prints the@@@ -9,3 +9,9 @@ description = "Sentinel provider key in the auth test that asserts secrets are ncondition = "AND"paths = ['''^test/auth\.test\.js$''']regexes = ['''^sk-test-THIS-MUST-NEVER-BE-RETURNED-0123456789$''']++++[[allowlists]]++description = "Made-up provider key in test/credentials.test.js (a person's own key is stored and used; the value is a placeholder, not a secret)"++condition = "AND"++paths = ['''^test/credentials\.test\.js$''']++regexes = ['''sk-test-dana-0123456789WXYZ''']@@@ -13,7 +13,7 @@"better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"},@@@ -1395,9 +1395,9 @@}},"node_modules/openvibe-contracts": {-- "version": "0.49.0",-- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",-- "integrity": "sha512-lapsLoBm3ZpndgiOMELdCBh2PJtuuyqj6SUhQsfwJkxnLdQfyhXfRFgiuLh+iQrg4kqtG9611191Mp8zAfk7fQ==",++ "version": "0.73.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0",++ "integrity": "sha512-SPgU/Ys9TLa8K+++V38C3uK50hNXXUeSPrIyWBT9HdY7eOsE/dnNRZqQnuUzihHOK22aQI15bRBE7u944tW0WA==","license": "AGPL-3.0-only","dependencies": {"ajv": "^8.17.1",@@@ -19,7 +19,7 @@"better-sqlite3": "^11.10.0","dotenv": "^16.4.7","express": "^4.21.2",-- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",++ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0","openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"},@@@ -12,7 +12,7 @@ const { createOps } = require('./ops');const { consoleRouter } = require('./console');const pkg = require('../package.json');--function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console }) {++function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', 'loopback');@@@ -46,6 +46,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys// Operator actions shared by the admin API and the operator console (one implementation, one audit row).const ops = createOps({ db, registry, pool, quotas, cache, runs });app.use(runsRouter({ runs, registry, auth, config, log }));++ if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));// The operator console (/console, /auth/*): Network staff, server-rendered, no scripts (server/console).app.use(consoleRouter({ config, db, registry, pool, quotas, cache, runs, keys, ops, env, clock, fetchImpl, log }));@@@ -74,6 +74,8 @@ function load(env = process.env) {networkUrl: trimUrl(env.OV_NETWORK_URL || 'https://openvibe.network'),networkInternalUrl: trimUrl(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000'),issuer: trimUrl(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),++ // A person's own provider keys (server/credentials.js): AES-256-GCM, 64 hex characters; unset = credentials off.++ credentialsKey: String(env.AI_CREDENTIALS_KEY || '').trim(),networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,audience: 'openvibe.ai',// Namespaces fail closed (server/auth.js): no `ns` claim, no namespaced run, except thennew file mode 100644@@@ -0,0 +1,158 @@++'use strict';++/**++ * A person's own provider key (roadmap WS-O task 2; Contracts 0.73.0 ai.credential.manage, ai.credential-put@1,++ * ai.credential@1). The service that holds the person's consent (Live, for a streamer's AI viewers) stores the++ * key here; it never leaves OpenVibe.AI again. Runs that name the credential (ai.run-request credential { subject })++ * call that provider with that key only (server/providers/index.js executeWithCredential), never a shared key.++ *++ * PUT /api/v1/credentials/:subject store or replace (the storing service is the owner)++ * GET /api/v1/credentials/:subject read it back without the key: a four-character hint, today's spend++ * DELETE /api/v1/credentials/:subject++ *++ * At rest the key is AES-256-GCM encrypted with AI_CREDENTIALS_KEY (64 hex characters), bound to its owner and++ * subject (the additional data), so a row copied to another owner or subject does not decrypt. Without the key++ * configured the routes answer 503 and credentialed runs are refused. Spend is what the credential's requests++ * cost today (provider_key byo:<owner>:<subject>), checked against the optional budget_usd_per_day.++ */++const crypto = require('crypto');++const express = require('express');++const { validate } = require('openvibe-contracts');++const { AiError } = require('./util');++++const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;++++function providerKeyFor(owner, subject) { return `byo:${owner}:${subject}`; }++++function createCredentials({ db, config, clock = { now: () => Date.now() } }) {++ db.exec(`CREATE TABLE IF NOT EXISTS subject_credentials (++ owner TEXT NOT NULL,++ subject TEXT NOT NULL,++ provider TEXT NOT NULL CHECK (provider IN ('openai', 'anthropic')),++ base_url TEXT,++ key_enc TEXT NOT NULL,++ key_hint TEXT NOT NULL,++ models TEXT NOT NULL DEFAULT '{}',++ budget_usd_day REAL,++ created_at TEXT NOT NULL,++ updated_at TEXT NOT NULL,++ last_used_at TEXT,++ PRIMARY KEY (owner, subject)++ )`);++ const iso = (ms) => new Date(ms).toISOString();++++ function secretKey() {++ const hex = String(config.credentialsKey || '');++ if (!/^[0-9a-f]{64}$/i.test(hex)) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service (AI_CREDENTIALS_KEY)');++ return Buffer.from(hex, 'hex');++ }++ function seal(plain, aad) {++ const iv = crypto.randomBytes(12);++ const c = crypto.createCipheriv('aes-256-gcm', secretKey(), iv);++ c.setAAD(Buffer.from(aad));++ const body = Buffer.concat([c.update(String(plain), 'utf8'), c.final()]);++ return `v1:${iv.toString('base64')}:${c.getAuthTag().toString('base64')}:${body.toString('base64')}`;++ }++ function open(sealed, aad) {++ const [v, iv, tag, body] = String(sealed).split(':');++ if (v !== 'v1') throw new Error('unknown credential format');++ const d = crypto.createDecipheriv('aes-256-gcm', secretKey(), Buffer.from(iv, 'base64'));++ d.setAAD(Buffer.from(aad));++ d.setAuthTag(Buffer.from(tag, 'base64'));++ return Buffer.concat([d.update(Buffer.from(body, 'base64')), d.final()]).toString('utf8');++ }++++ function spentToday(owner, subject) {++ const day = iso(clock.now()).slice(0, 10);++ const r = db.prepare("SELECT COALESCE(SUM(cost_usd), 0) AS c FROM requests WHERE provider_key = ? AND status = 'ok' AND created_at >= ?").get(providerKeyFor(owner, subject), `${day}T00:00:00.000Z`);++ return r ? Number(r.c) || 0 : 0;++ }++++ function view(row) {++ return {++ subject: row.subject, owner: row.owner, provider: row.provider, base_url: row.base_url || null, key_hint: row.key_hint,++ models: JSON.parse(row.models || '{}'), budget_usd_per_day: row.budget_usd_day == null ? null : row.budget_usd_day,++ spent_usd_today: Math.round(spentToday(row.owner, row.subject) * 1e6) / 1e6,++ created_at: row.created_at, updated_at: row.updated_at, last_used_at: row.last_used_at || null,++ };++ }++ const row = (owner, subject) => db.prepare('SELECT * FROM subject_credentials WHERE owner = ? AND subject = ?').get(owner, subject);++++ function put(owner, subject, body) {++ if (!SUBJECT_RE.test(String(subject))) throw new AiError(400, 'input.invalid', 'subject must be a usr_ subject');++ const v = validate('ai.credential-put@1', body);++ if (!v.valid) throw new AiError(422, 'input.invalid', 'body does not match ai.credential-put@1', { errors: v.errors });++ const at = iso(clock.now());++ const key = String(body.api_key);++ const prev = row(owner, subject);++ db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, budget_usd_day, created_at, updated_at)++ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)++ ON CONFLICT(owner, subject) DO UPDATE SET provider = excluded.provider, base_url = excluded.base_url, key_enc = excluded.key_enc,++ key_hint = excluded.key_hint, models = excluded.models, budget_usd_day = excluded.budget_usd_day, updated_at = excluded.updated_at`)++ .run(owner, subject, body.provider, body.base_url || null, seal(key, `${owner}|${subject}`), `…${key.slice(-4)}`, JSON.stringify(body.models || {}),++ body.budget_usd_per_day ? Number(body.budget_usd_per_day) : null, prev ? prev.created_at : at, at);++ return view(row(owner, subject));++ }++++ function get(owner, subject) {++ const r = row(owner, subject);++ if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');++ return view(r);++ }++++ function remove(owner, subject) {++ return db.prepare('DELETE FROM subject_credentials WHERE owner = ? AND subject = ?').run(owner, subject).changes > 0;++ }++++ /** Checks a run may use the credential (exists, within budget); throws AiError otherwise. */++ function admit(owner, subject) {++ const r = row(owner, subject);++ if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');++ secretKey();++ if (r.budget_usd_day != null && r.budget_usd_day > 0 && spentToday(owner, subject) >= r.budget_usd_day) {++ const tomorrow = new Date(iso(clock.now()).slice(0, 10) + 'T00:00:00.000Z').getTime() + 86400000;++ throw new AiError(429, 'quota.exceeded', "this person's own daily budget for their key is spent", { retry_after_seconds: Math.max(60, Math.ceil((tomorrow - clock.now()) / 1000)), quota: { scope: 'credential' } });++ }++ }++++ /** The decrypted credential for one execution: { providerKey, provider, base_url, apiKey, models }. */++ function forRun(owner, subject) {++ const r = row(owner, subject);++ if (!r) throw new AiError(404, 'credential.not_found', 'the credential was deleted');++ db.prepare('UPDATE subject_credentials SET last_used_at = ? WHERE owner = ? AND subject = ?').run(iso(clock.now()), owner, subject);++ return { providerKey: providerKeyFor(owner, subject), provider: r.provider, base_url: r.base_url || null, apiKey: open(r.key_enc, `${owner}|${subject}`), models: JSON.parse(r.models || '{}') };++ }++++ return { put, get, remove, admit, forRun, spentToday, providerKeyFor };++}++++/** PUT/GET/DELETE /api/v1/credentials/:subject (ai.credential.manage; a service token speaks for itself). */++function credentialsRouter({ credentials, auth, registry, sendError }) {++ const r = express.Router();++ const guard = auth.requireCap('ai.credential.manage');++ const owner = (req) => {++ const s = req.principal && req.principal.subject;++ if (!s || s.type !== 'service') throw new AiError(403, 'capability.denied', 'only a service stores credentials for the people it serves');++ return s.id;++ };++ r.put('/api/v1/credentials/:subject', guard, (req, res) => {++ try {++ const out = credentials.put(owner(req), req.params.subject, req.body || {});++ registry.audit(req.principal.sub, 'credential.put', 'credential', `${out.owner}:${out.subject}`, { trace: req.ov && req.ov.traceId, metadata: { provider: out.provider, base_url: out.base_url } });++ res.set('Cache-Control', 'no-store').json(out);++ } catch (err) { sendError(res, err, req.ov); }++ });++ r.get('/api/v1/credentials/:subject', guard, (req, res) => {++ try { res.set('Cache-Control', 'no-store').json(credentials.get(owner(req), req.params.subject)); } catch (err) { sendError(res, err, req.ov); }++ });++ r.delete('/api/v1/credentials/:subject', guard, (req, res) => {++ try {++ const o = owner(req);++ const gone = credentials.remove(o, req.params.subject);++ if (gone) registry.audit(req.principal.sub, 'credential.delete', 'credential', `${o}:${req.params.subject}`, { trace: req.ov && req.ov.traceId });++ res.status(gone ? 204 : 404).end();++ } catch (err) { sendError(res, err, req.ov); }++ });++ return r;++}++++module.exports = { createCredentials, credentialsRouter, providerKeyFor };@@@ -21,14 +21,14 @@ const { createKeyStore, createAuth } = require('./auth');const schemas = require('./schemas');const { createApp } = require('./app');--async function start({ config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true } = {}) {++async function start({ config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true, credentialFetch = null } = {}) {config = config || load(env);schemas.configure(config.schemaCache);const db = openDb(config.dbPath);const registry = createRegistry(db, { clock, env });const quotas = createQuotas(db, { clock, registry });const cache = createCache(db, { clock });-- const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log });++ const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch });const fetcher = createFetcher(config);const engine = createEngine({ registry, pool, fetcher, quotas, config, log });// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.@@@ -36,14 +36,16 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl// ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });userModules.ensureSchema();-- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules });++ // A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.++ const credentials = require('./credentials').createCredentials({ db, config, clock });++ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials });seed({ registry, quotas, config, env, db });const interrupted = runs.recoverInterrupted();if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });const auth = createAuth({ config, keys });-- const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log });++ const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials });const keyLoaded = keys.start().catch(() => null);const housekeeping = setInterval(() => {nnew file mode 100644@@@ -0,0 +1,51 @@++'use strict';++/**++ * fetch() for a base URL a person chose (their own provider key, server/credentials.js): https only, no++ * credentials in the URL, no IP-literal hosts, and the connection goes only to public unicast addresses, checked++ * at connect time (fetcher.js safeLookup), so a DNS answer cannot point it inside the network. No redirects; a++ * response is capped at 8 MB. Returns a standard Response, which is all the provider adapters use.++ */++const https = require('https');++const net = require('net');++const { safeLookup } = require('../fetcher');++++const MAX_BYTES = 8 * 1024 * 1024;++++function refuse(message) {++ const e = new Error(message);++ e.code = 'EADDRNOTPUBLIC';++ return e;++}++++function guardedFetch(input, init = {}) {++ return new Promise((resolve, reject) => {++ let url;++ try { url = new URL(String(input)); } catch { return reject(refuse('not a valid URL')); }++ if (url.protocol !== 'https:') return reject(refuse('only https endpoints are called with a person\'s own key'));++ if (url.username || url.password) return reject(refuse('credentials in URLs are not allowed'));++ if (net.isIP(url.hostname.replace(/^\[|\]$/g, ''))) return reject(refuse('IP-literal hosts are not called'));++ if (typeof init.body !== 'string' && init.body != null) return reject(refuse('only JSON bodies are sent with a person\'s own key'));++ const headers = { ...(init.headers || {}) };++ if (init.body != null) headers['Content-Length'] = Buffer.byteLength(init.body);++ const req = https.request(url, { method: init.method || 'GET', headers, lookup: safeLookup });++ const signal = init.signal;++ const onAbort = () => { req.destroy(); reject(Object.assign(new Error('aborted'), { name: 'AbortError' })); };++ if (signal) { if (signal.aborted) return onAbort(); signal.addEventListener('abort', onAbort, { once: true }); }++ req.on('error', (err) => { if (signal) signal.removeEventListener('abort', onAbort); reject(err); });++ req.on('response', (res) => {++ const chunks = []; let bytes = 0;++ res.on('data', (c) => { bytes += c.length; if (bytes > MAX_BYTES) { res.destroy(); req.destroy(); reject(new Error('response too large')); } else chunks.push(c); });++ res.on('error', reject);++ res.on('end', () => {++ if (signal) signal.removeEventListener('abort', onAbort);++ const h = new Headers();++ for (const [k, v] of Object.entries(res.headers)) if (v != null) h.set(k, Array.isArray(v) ? v.join(', ') : String(v));++ resolve(new Response(res.statusCode === 204 ? null : Buffer.concat(chunks), { status: res.statusCode, headers: h }));++ });++ });++ if (init.body != null) req.write(init.body);++ req.end();++ });++}++++module.exports = { guardedFetch, MAX_BYTES };@@@ -29,7 +29,7 @@ function callerFault(err) {return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);}--function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console }) {++function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null }) {const cache = new Map(); // key -> { stamp, adapter }const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)@@@ -132,7 +132,58 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no* ctx: { runId, routeKey, routeVersion, signal, logRequest(entry), debugRaw }* req: canonical request (system, messages, image, json, maxTokens, temperature, cacheKey, input, filePath, ...)*/++ /**++ * A run with a person's own key (server/credentials.js): that provider only, no fallback, no circuit breaker++ * shared with anyone, chat operations only; the base URL they chose is reached through guardedFetch. One++ * request-log row per attempt under provider_key byo:<owner>:<subject>, priced by the model's list price.++ */++ async function executeWithCredential(cred, operation, req, ctx = {}) {++ if (operation !== 'chat') throw new AiError(422, 'credential.unsupported', `a person's own key is used for chat only, not ${operation}`);++ const guardedFetch = credentialFetch || require('./guarded-fetch').guardedFetch;++ const role = ctx.role && cred.models[ctx.role] ? ctx.role : null;++ const model = (req.image && cred.models.vision) || (role && cred.models[role]) || cred.models.chat || (cred.provider === 'anthropic' ? 'claude-haiku-4-5-20251001' : 'gpt-4o-mini');++ const record = { key: cred.providerKey, kind: cred.provider, base_url: cred.base_url, capabilities: ['chat', 'json', 'vision'], timeout_ms: 60000 };++ const a = cred.provider === 'anthropic' ? createAnthropicProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch }) : createOpenAiProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch });++ const base = { operation, provider_key: cred.providerKey, model_key: model, route_key: ctx.routeKey || null, route_version: ctx.routeVersion || null, fallback: 0, prompt_hash: ctx.promptHash || null };++ const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || 60000, 120000));++ let lastErr = null;++ for (let attempt = 0; attempt <= config.providerRetries; attempt++) {++ if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');++ const t0 = Date.now();++ try {++ const result = await a.chat({ ...req, model, signal: ctx.signal, timeoutMs });++ const usage = result.usage || { input: 0, output: 0, cached: 0 };++ const listPrice = (() => {++ const table = config.pricing.table || {};++ const m = String(result.model || model).toLowerCase();++ let best = null; let len = -1;++ for (const [k, v] of Object.entries(table)) { const key = String(k).toLowerCase(); if (key !== 'default' && m.startsWith(key) && key.length > len && v && typeof v === 'object') { best = v; len = key.length; } }++ const b = best || table.default || {};++ const inRate = Number(b.in) || 0;++ return { in: inRate, out: Number(b.out) || 0, cached: b.cached != null ? Number(b.cached) : inRate * 0.1 };++ })();++ const input = Math.max(0, (usage.input || 0) - (usage.cached || 0));++ const cost = (input / 1e6) * listPrice.in + ((usage.cached || 0) / 1e6) * listPrice.cached + ((usage.output || 0) / 1e6) * listPrice.out;++ const latency = Date.now() - t0;++ if (ctx.logRequest) {++ ctx.logRequest({ ...base, model_key: result.model || model, status: 'ok', output_hash: sha256(result.json || result.text || ''), tokens_in: usage.input || 0, tokens_out: usage.output || 0,++ tokens_cached: usage.cached || 0, tokens_estimated: usage.estimated ? 1 : 0, cost_usd: cost, latency_ms: latency });++ }++ return { result, provider: cred.providerKey, providerKind: cred.provider, model: result.model || model, fallbackUsed: false, usage, cost, latencyMs: latency, synthetic: false, tried: [], startedAt: t0 };++ } catch (err) {++ lastErr = err;++ if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');++ if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });++ if (err && err.code === 'EADDRNOTPUBLIC') throw new AiError(422, 'credential.endpoint_refused', err.message);++ if (attempt < config.providerRetries && retryable(err)) { await sleep(config.retryDelayMs, ctx.signal).catch(() => {}); continue; }++ break;++ }++ }++ throw new AiError(502, 'credential.provider_failed', `the person's own provider did not answer: ${String(lastErr && lastErr.message || lastErr).slice(0, 200)}`);++ }++async function execute(route, operation, req, ctx = {}) {++ if (ctx.credential) return executeWithCredential(ctx.credential, operation, req, ctx);const features = [operation];if (req.image) features.push('vision');if (req.json) features.push('json');@@@ -201,7 +252,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.nothrow new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried });}-- return { adapter, health, resetHealth, execute, priceFor, costOf, stats, statFor, candidates };++ return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, stats, statFor, candidates };}module.exports = { createProviderPool };@@@ -58,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {return { value, full: !redacted };}--function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null }) {++function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null }) {const inflight = new Map(); // run id -> { controller, promise, release, caller }const queue = []; // run ids waiting for the workerlet active = 0;@@@ -168,6 +168,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowconst routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;const route = routeKey && !String(routeKey).includes('{') ? registry.resolveRoute(routeKey) : null;const options = body.options && typeof body.options === 'object' ? body.options : {};++ // A person's own key (WS-O task 2): only the service that stored it, within its daily budget; never cached.++ const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;++ if (credentialSubject !== null) {++ if (!credentials) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service');++ if (requester.type !== 'service') throw new AiError(403, 'capability.denied', 'only the service that stored a credential runs with it');++ credentials.admit(requester.id, credentialSubject);++ }const ctx = {requesterType: requester.type, requesterId: requester.id,actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,@@@ -184,12 +191,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowsource_service: requester.type === 'service' ? requester.id : null,target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,-- options: JSON.stringify({ cache: options.cache !== false, debug: Boolean(options.debug), input_retained: kept.full }), created_at: now,++ options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null, debug: Boolean(options.debug), input_retained: kept.full, ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}) }), created_at: now,};base.input = stableStringify(kept.value);// Cache (scoped; never crosses requester/actor/target/attribution)-- const scope = options.cache === false ? null : cache.scopeFor(wf.cache_mode, ctx);++ const scope = options.cache === false || credentialSubject !== null ? null : cache.scopeFor(wf.cache_mode, ctx);let cacheKey = null;if (scope) {cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });@@@ -215,7 +222,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now// Queue caps, then quota: both before any provider call (a refused run is not counted).const caller = `${requester.type}:${requester.id}`;admit(caller);-- const reserved = quotas.reserve(ctx);++ // A run with a person's own key spends their money, not the shared quotas (its own budget was checked above).++ const reserved = credentialSubject !== null ? null : quotas.reserve(ctx);db.transaction(() => {db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,@@@ -284,7 +292,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (preferences.history === false) {db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = json_set(options, '$.input_retained', json('false'), '$.history', json('false')) WHERE id = ?").run(id);}-- const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences });++ const credential = opts.credential_subject ? credentials.forRun(row.requester_id, opts.credential_subject) : null;++ const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences,++ ...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}) });if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');// The status, its citations and the ai.run.succeeded event commit together.const done = db.transaction(() => {@@@ -298,7 +308,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowreturn d;})();if (!done.changes) return; // cancelled while finishing-- if (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });++ if (r.provider && reserved) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {@@@ -317,7 +327,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { nowif (f.changes) events.runChanged(getRow(id));})();const spent = db.prepare("SELECT COALESCE(SUM(tokens_in),0) ti, COALESCE(SUM(tokens_out),0) tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);-- if (spent && (spent.ti || spent.tout)) quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });++ if (reserved && spent && (spent.ti || spent.tout)) quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });} finally {inflight.delete(id);}nnew file mode 100644@@@ -0,0 +1,99 @@++'use strict';++// A person's own provider key (roadmap WS-O task 2; Contracts 0.73.0 ai.credential.manage):++// - stored by a service holding ai.credential.manage, encrypted at rest and bound to its owner and subject, never++// returned (a four-character hint only); another service neither sees nor uses it; off without AI_CREDENTIALS_KEY;++// - a run naming it calls that provider with that key only (no fallback, never cached, the shared quota untouched),++// logged under byo:<owner>:<subject>, and its own daily budget stops further runs (429);++// - the guarded fetch reaches only public https endpoints.++const assert = require('assert');++const { boot, request, token, suite, ALL } = require('./helpers');++const { guardedFetch } = require('../server/providers/guarded-fetch');++++const t = suite('credentials');++const KEY = 'ab'.repeat(32);++const DANA = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP1';++const API_KEY = 'sk-test-dana-0123456789WXYZ';++const live = token('live', [...ALL, 'ai.credential.manage']);++const tools = token('tools', [...ALL, 'ai.credential.manage']);++const noCap = token('live', ALL);++const calls = [];++// The person's provider: an OpenAI-compatible endpoint, answered in-process (the guarded fetch is tested below).++const credentialFetch = async (url, init) => {++ calls.push({ url, auth: init.headers.Authorization || init.headers.authorization, body: JSON.parse(init.body) });++ return new Response(JSON.stringify({ model: 'gpt-4o-mini', choices: [{ message: { role: 'assistant', content: 'hello from dana\'s key' } }], usage: { prompt_tokens: 1000, completion_tokens: 1000 } }), { status: 200, headers: { 'content-type': 'application/json' } });++};++let h;++++t.test('boot with a credentials key', async () => {++ h = await boot({ env: { AI_CREDENTIALS_KEY: KEY, AI_PRICING_JSON: JSON.stringify({ 'gpt-4o-mini': { in: 1, out: 1 } }) }, credentialFetch });++});++++t.test('store, read without the key, refuse other services', async () => {++ const put = { provider: 'openai', base_url: 'https://llm.example.com/v1', api_key: API_KEY, models: { chat: 'gpt-4o-mini' }, budget_usd_per_day: 0.003 };++ let r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: noCap, body: put });++ assert.strictEqual(r.status, 403, 'needs ai.credential.manage');++ r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { ...put, base_url: 'http://10.0.0.1/v1' } });++ assert.strictEqual(r.status, 422, 'https only');++ r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: put });++ assert.strictEqual(r.status, 200, r.text);++ assert.deepStrictEqual([r.body.owner, r.body.key_hint, r.body.provider], ['live', '…WXYZ', 'openai']);++ assert.ok(!r.text.includes(API_KEY), 'the key never comes back');++ const { validate } = require('openvibe-contracts');++ assert.ok(validate('ai.credential@1', r.body).valid, JSON.stringify(validate('ai.credential@1', r.body).errors));++ const row = h.db.prepare('SELECT key_enc FROM subject_credentials WHERE owner = ? AND subject = ?').get('live', DANA);++ assert.ok(row.key_enc.startsWith('v1:') && !row.key_enc.includes(API_KEY), 'encrypted at rest');++ r = await request(h.base, 'GET', `/api/v1/credentials/${DANA}`, { tok: tools });++ assert.strictEqual(r.status, 404, 'another service does not see it');++ // A row moved to another owner does not decrypt: the key is bound to owner and subject.++ h.db.prepare("INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, created_at, updated_at) SELECT 'tools', subject, provider, base_url, key_enc, key_hint, models, created_at, updated_at FROM subject_credentials WHERE owner = 'live'").run();++ const creds = require('../server/credentials').createCredentials({ db: h.db, config: h.config });++ assert.throws(() => creds.forRun('tools', DANA));++ h.db.prepare("DELETE FROM subject_credentials WHERE owner = 'tools'").run();++});++++t.test('a run with the credential uses that key only, uncached, and its budget stops it', async () => {++ const run = (tok = live) => request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'live.viewers.line', input: { role: 'chat', user: 'say hi' }, credential: { subject: DANA } } });++ let r = await run();++ assert.strictEqual(r.status, 201, r.text);++ assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run));++ assert.strictEqual(r.body.run.output.text, "hello from dana's key");++ assert.strictEqual(calls.length, 1);++ assert.strictEqual(calls[0].url, 'https://llm.example.com/v1/chat/completions');++ assert.strictEqual(calls[0].auth, `Bearer ${API_KEY}`);++ const reqs = h.db.prepare('SELECT provider_key, status, cost_usd FROM requests WHERE run_id = ?').all(r.body.run.id);++ assert.deepStrictEqual(reqs.map((x) => [x.provider_key, x.status]), [[`byo:live:${DANA}`, 'ok']]);++ assert.ok(Math.abs(reqs[0].cost_usd - 0.002) < 1e-9, 'priced at list price');++ r = await run();++ assert.strictEqual(r.body.run.status, 'succeeded');++ assert.strictEqual(calls.length, 2, 'never served from the cache');++ r = await run();++ assert.deepStrictEqual([r.status, r.body.code], [429, 'quota.exceeded'], 'the daily budget (0.003) is spent');++ assert.strictEqual(calls.length, 2);++ r = await request(h.base, 'GET', `/api/v1/credentials/${DANA}`, { tok: live });++ assert.ok(Math.abs(r.body.spent_usd_today - 0.004) < 1e-9);++ // Another service cannot run with it; neither can a person's token.++ r = await run(tools);++ assert.deepStrictEqual([r.status, r.body.code], [404, 'credential.not_found']);++ r = await run(token('dana', ALL, { sub: DANA, actorType: 'user' }));++ assert.ok([401, 403].includes(r.status), `a user token: ${r.status}`);++ // Deleted: runs refuse it.++ assert.strictEqual((await request(h.base, 'DELETE', `/api/v1/credentials/${DANA}`, { tok: live })).status, 204);++ r = await run();++ assert.deepStrictEqual([r.status, r.body.code], [404, 'credential.not_found']);++});++++t.test('the guarded fetch reaches only public https endpoints', async () => {++ for (const [url, why] of [['http://example.com/v1', 'https only'], ['https://127.0.0.1/v1', 'IP literal'], ['https://user:[email protected]/v1', 'credentials in URL'], ['https://localhost:9/v1', 'resolves to loopback']]) {++ await assert.rejects(guardedFetch(url, { method: 'POST', headers: {}, body: '{}' }), (e) => e.code === 'EADDRNOTPUBLIC', why);++ }++});++++t.test('without AI_CREDENTIALS_KEY the routes answer 503', async () => {++ const off = await boot({});++ const r = await request(off.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { provider: 'openai', api_key: API_KEY } });++ assert.deepStrictEqual([r.status, r.body.code], [503, 'credentials.unavailable']);++ await off.stop();++ await h.stop();++});++++t.run();@@@ -42,14 +42,14 @@ function tmpDir() {}/** Boot the service. `env` goes through config.load() and is also what secrets resolve against. */--async function boot({ env = {}, dir = tmpDir(), clock, fetchImpl } = {}) {++async function boot({ env = {}, dir = tmpDir(), clock, fetchImpl, credentialFetch } = {}) {const fullEnv = {NODE_ENV: 'test', PORT: '0', AI_DB_PATH: path.join(dir, 'ai.db'), OV_NETWORK_PUBLIC_KEY: publicKey,AI_PROVIDER_RETRY_DELAY_MS: '5', AI_QUOTA_SERVICE_RPM: '0', AI_QUOTA_SERVICE_RPD: '0',...env,};const config = load(fullEnv);-- const h = await start({ config, env: fullEnv, log: silent, clock, fetchImpl });++ const h = await start({ config, env: fullEnv, log: silent, clock, fetchImpl, credentialFetch });const base = `http://127.0.0.1:${h.server.address().port}`;return { ...h, base, dir, env: fullEnv, stop: () => h.close() };}
@@ -39,6 +39,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
@@ -39,6 +39,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
# Audience a Network user token must carry
# Audience a Network user token must carry
#AI_SSO_AUDIENCE=openvibe.network
#AI_SSO_AUDIENCE=openvibe.network
+# ── A person's own provider keys (server/credentials.js, WS-O task 2) ──
+# 64 hex characters (openssl rand -hex 32). Keys stored through PUT /api/v1/credentials/:subject are AES-256-GCM
+# encrypted with it; unset turns the credentials routes and credentialed runs off (503). Changing it makes every
+# stored key unreadable: the storing service then has to send them again.
+#AI_CREDENTIALS_KEY=
+
# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──
# ── The shared provider — the same names as OpenVibe.Live's admin AI settings ──
# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_model_<role>, ai_pricing_json,
# (ai_enabled, ai_provider, ai_base_url, ai_api_key, ai_model, ai_model_<role>, ai_pricing_json,
# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day). scripts/import-from-live.js prints the
# ai_input/output_cost_per_mtok, ai_max_cost_usd_per_day). scripts/import-from-live.js prints the
@@ -9,3 +9,9 @@ description = "Sentinel provider key in the auth test that asserts secrets are n
@@ -9,3 +9,9 @@ description = "Sentinel provider key in the auth test that asserts secrets are n
condition = "AND"
condition = "AND"
paths = ['''^test/auth\.test\.js$''']
paths = ['''^test/auth\.test\.js$''']
regexes = ['''^sk-test-THIS-MUST-NEVER-BE-RETURNED-0123456789$''']
regexes = ['''^sk-test-THIS-MUST-NEVER-BE-RETURNED-0123456789$''']
+
+[[allowlists]]
+description = "Made-up provider key in test/credentials.test.js (a person's own key is stored and used; the value is a placeholder, not a secret)"
+condition = "AND"
+paths = ['''^test/credentials\.test\.js$''']
+regexes = ['''sk-test-dana-0123456789WXYZ''']
@@ -13,7 +13,7 @@
@@ -13,7 +13,7 @@
"better-sqlite3": "^11.10.0",
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
},
@@ -1395,9 +1395,9 @@
@@ -1395,9 +1395,9 @@
}
}
},
},
"node_modules/openvibe-contracts": {
"node_modules/openvibe-contracts": {
- "version": "0.49.0",
- "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
- "integrity": "sha512-lapsLoBm3ZpndgiOMELdCBh2PJtuuyqj6SUhQsfwJkxnLdQfyhXfRFgiuLh+iQrg4kqtG9611191Mp8zAfk7fQ==",
+ "version": "0.73.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0",
+ "integrity": "sha512-SPgU/Ys9TLa8K+++V38C3uK50hNXXUeSPrIyWBT9HdY7eOsE/dnNRZqQnuUzihHOK22aQI15bRBE7u944tW0WA==",
"license": "AGPL-3.0-only",
"license": "AGPL-3.0-only",
"dependencies": {
"dependencies": {
"ajv": "^8.17.1",
"ajv": "^8.17.1",
@@ -19,7 +19,7 @@
@@ -19,7 +19,7 @@
"better-sqlite3": "^11.10.0",
"better-sqlite3": "^11.10.0",
"dotenv": "^16.4.7",
"dotenv": "^16.4.7",
"express": "^4.21.2",
"express": "^4.21.2",
- "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
+ "openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.73.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.11.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
},
@@ -12,7 +12,7 @@ const { createOps } = require('./ops');
@@ -12,7 +12,7 @@ const { createOps } = require('./ops');
const { consoleRouter } = require('./console');
const { consoleRouter } = require('./console');
const pkg = require('../package.json');
const pkg = require('../package.json');
-function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console }) {
+function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env = process.env, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, log = console, credentials = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
app.set('trust proxy', 'loopback');
@@ -46,6 +46,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
@@ -46,6 +46,7 @@ function createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys
// Operator actions shared by the admin API and the operator console (one implementation, one audit row).
// Operator actions shared by the admin API and the operator console (one implementation, one audit row).
const ops = createOps({ db, registry, pool, quotas, cache, runs });
const ops = createOps({ db, registry, pool, quotas, cache, runs });
app.use(runsRouter({ runs, registry, auth, config, log }));
app.use(runsRouter({ runs, registry, auth, config, log }));
+ if (credentials) app.use(require('./credentials').credentialsRouter({ credentials, auth, registry, sendError: (res, err, ctx) => require('./api/runs').sendError(res, err, ctx, log) }));
app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));
app.use(adminRouter({ db, registry, pool, quotas, cache, runs, auth, ops, log }));
// The operator console (/console, /auth/*): Network staff, server-rendered, no scripts (server/console).
// The operator console (/console, /auth/*): Network staff, server-rendered, no scripts (server/console).
app.use(consoleRouter({ config, db, registry, pool, quotas, cache, runs, keys, ops, env, clock, fetchImpl, log }));
app.use(consoleRouter({ config, db, registry, pool, quotas, cache, runs, keys, ops, env, clock, fetchImpl, log }));
@@ -74,6 +74,8 @@ function load(env = process.env) {
@@ -74,6 +74,8 @@ function load(env = process.env) {
networkUrl: trimUrl(env.OV_NETWORK_URL || 'https://openvibe.network'),
networkUrl: trimUrl(env.OV_NETWORK_URL || 'https://openvibe.network'),
networkInternalUrl: trimUrl(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000'),
networkInternalUrl: trimUrl(env.OV_NETWORK_INTERNAL_URL || 'http://127.0.0.1:4000'),
issuer: trimUrl(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),
issuer: trimUrl(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),
+ // A person's own provider keys (server/credentials.js): AES-256-GCM, 64 hex characters; unset = credentials off.
+ credentialsKey: String(env.AI_CREDENTIALS_KEY || '').trim(),
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
audience: 'openvibe.ai',
audience: 'openvibe.ai',
// Namespaces fail closed (server/auth.js): no `ns` claim, no namespaced run, except the
// Namespaces fail closed (server/auth.js): no `ns` claim, no namespaced run, except the
new file mode 100644
new file mode 100644
@@ -0,0 +1,158 @@
@@ -0,0 +1,158 @@
+'use strict';
+/**
+ * A person's own provider key (roadmap WS-O task 2; Contracts 0.73.0 ai.credential.manage, ai.credential-put@1,
+ * ai.credential@1). The service that holds the person's consent (Live, for a streamer's AI viewers) stores the
+ * key here; it never leaves OpenVibe.AI again. Runs that name the credential (ai.run-request credential { subject })
+ * call that provider with that key only (server/providers/index.js executeWithCredential), never a shared key.
+ *
+ * PUT /api/v1/credentials/:subject store or replace (the storing service is the owner)
+ * GET /api/v1/credentials/:subject read it back without the key: a four-character hint, today's spend
+ * DELETE /api/v1/credentials/:subject
+ *
+ * At rest the key is AES-256-GCM encrypted with AI_CREDENTIALS_KEY (64 hex characters), bound to its owner and
+ * subject (the additional data), so a row copied to another owner or subject does not decrypt. Without the key
+ * configured the routes answer 503 and credentialed runs are refused. Spend is what the credential's requests
+ * cost today (provider_key byo:<owner>:<subject>), checked against the optional budget_usd_per_day.
+ */
+const crypto = require('crypto');
+const express = require('express');
+const { validate } = require('openvibe-contracts');
+const { AiError } = require('./util');
+
+const SUBJECT_RE = /^usr_[0-9A-HJKMNP-TV-Z]{26}$/;
+
+function providerKeyFor(owner, subject) { return `byo:${owner}:${subject}`; }
+
+function createCredentials({ db, config, clock = { now: () => Date.now() } }) {
+ db.exec(`CREATE TABLE IF NOT EXISTS subject_credentials (
+ owner TEXT NOT NULL,
+ subject TEXT NOT NULL,
+ provider TEXT NOT NULL CHECK (provider IN ('openai', 'anthropic')),
+ base_url TEXT,
+ key_enc TEXT NOT NULL,
+ key_hint TEXT NOT NULL,
+ models TEXT NOT NULL DEFAULT '{}',
+ budget_usd_day REAL,
+ created_at TEXT NOT NULL,
+ updated_at TEXT NOT NULL,
+ last_used_at TEXT,
+ PRIMARY KEY (owner, subject)
+ )`);
+ const iso = (ms) => new Date(ms).toISOString();
+
+ function secretKey() {
+ const hex = String(config.credentialsKey || '');
+ if (!/^[0-9a-f]{64}$/i.test(hex)) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service (AI_CREDENTIALS_KEY)');
+ return Buffer.from(hex, 'hex');
+ }
+ function seal(plain, aad) {
+ const iv = crypto.randomBytes(12);
+ const c = crypto.createCipheriv('aes-256-gcm', secretKey(), iv);
+ c.setAAD(Buffer.from(aad));
+ const body = Buffer.concat([c.update(String(plain), 'utf8'), c.final()]);
+ return `v1:${iv.toString('base64')}:${c.getAuthTag().toString('base64')}:${body.toString('base64')}`;
+ }
+ function open(sealed, aad) {
+ const [v, iv, tag, body] = String(sealed).split(':');
+ if (v !== 'v1') throw new Error('unknown credential format');
+ const d = crypto.createDecipheriv('aes-256-gcm', secretKey(), Buffer.from(iv, 'base64'));
+ d.setAAD(Buffer.from(aad));
+ d.setAuthTag(Buffer.from(tag, 'base64'));
+ return Buffer.concat([d.update(Buffer.from(body, 'base64')), d.final()]).toString('utf8');
+ }
+
+ function spentToday(owner, subject) {
+ const day = iso(clock.now()).slice(0, 10);
+ const r = db.prepare("SELECT COALESCE(SUM(cost_usd), 0) AS c FROM requests WHERE provider_key = ? AND status = 'ok' AND created_at >= ?").get(providerKeyFor(owner, subject), `${day}T00:00:00.000Z`);
+ return r ? Number(r.c) || 0 : 0;
+ }
+
+ function view(row) {
+ return {
+ subject: row.subject, owner: row.owner, provider: row.provider, base_url: row.base_url || null, key_hint: row.key_hint,
+ models: JSON.parse(row.models || '{}'), budget_usd_per_day: row.budget_usd_day == null ? null : row.budget_usd_day,
+ spent_usd_today: Math.round(spentToday(row.owner, row.subject) * 1e6) / 1e6,
+ created_at: row.created_at, updated_at: row.updated_at, last_used_at: row.last_used_at || null,
+ };
+ }
+ const row = (owner, subject) => db.prepare('SELECT * FROM subject_credentials WHERE owner = ? AND subject = ?').get(owner, subject);
+
+ function put(owner, subject, body) {
+ if (!SUBJECT_RE.test(String(subject))) throw new AiError(400, 'input.invalid', 'subject must be a usr_ subject');
+ const v = validate('ai.credential-put@1', body);
+ if (!v.valid) throw new AiError(422, 'input.invalid', 'body does not match ai.credential-put@1', { errors: v.errors });
+ const at = iso(clock.now());
+ const key = String(body.api_key);
+ const prev = row(owner, subject);
+ db.prepare(`INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, budget_usd_day, created_at, updated_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ ON CONFLICT(owner, subject) DO UPDATE SET provider = excluded.provider, base_url = excluded.base_url, key_enc = excluded.key_enc,
+ key_hint = excluded.key_hint, models = excluded.models, budget_usd_day = excluded.budget_usd_day, updated_at = excluded.updated_at`)
+ .run(owner, subject, body.provider, body.base_url || null, seal(key, `${owner}|${subject}`), `…${key.slice(-4)}`, JSON.stringify(body.models || {}),
+ body.budget_usd_per_day ? Number(body.budget_usd_per_day) : null, prev ? prev.created_at : at, at);
+ return view(row(owner, subject));
+ }
+
+ function get(owner, subject) {
+ const r = row(owner, subject);
+ if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');
+ return view(r);
+ }
+
+ function remove(owner, subject) {
+ return db.prepare('DELETE FROM subject_credentials WHERE owner = ? AND subject = ?').run(owner, subject).changes > 0;
+ }
+
+ /** Checks a run may use the credential (exists, within budget); throws AiError otherwise. */
+ function admit(owner, subject) {
+ const r = row(owner, subject);
+ if (!r) throw new AiError(404, 'credential.not_found', 'no credential stored for that subject');
+ secretKey();
+ if (r.budget_usd_day != null && r.budget_usd_day > 0 && spentToday(owner, subject) >= r.budget_usd_day) {
+ const tomorrow = new Date(iso(clock.now()).slice(0, 10) + 'T00:00:00.000Z').getTime() + 86400000;
+ throw new AiError(429, 'quota.exceeded', "this person's own daily budget for their key is spent", { retry_after_seconds: Math.max(60, Math.ceil((tomorrow - clock.now()) / 1000)), quota: { scope: 'credential' } });
+ }
+ }
+
+ /** The decrypted credential for one execution: { providerKey, provider, base_url, apiKey, models }. */
+ function forRun(owner, subject) {
+ const r = row(owner, subject);
+ if (!r) throw new AiError(404, 'credential.not_found', 'the credential was deleted');
+ db.prepare('UPDATE subject_credentials SET last_used_at = ? WHERE owner = ? AND subject = ?').run(iso(clock.now()), owner, subject);
+ return { providerKey: providerKeyFor(owner, subject), provider: r.provider, base_url: r.base_url || null, apiKey: open(r.key_enc, `${owner}|${subject}`), models: JSON.parse(r.models || '{}') };
+ }
+
+ return { put, get, remove, admit, forRun, spentToday, providerKeyFor };
+}
+
+/** PUT/GET/DELETE /api/v1/credentials/:subject (ai.credential.manage; a service token speaks for itself). */
+function credentialsRouter({ credentials, auth, registry, sendError }) {
+ const r = express.Router();
+ const guard = auth.requireCap('ai.credential.manage');
+ const owner = (req) => {
+ const s = req.principal && req.principal.subject;
+ if (!s || s.type !== 'service') throw new AiError(403, 'capability.denied', 'only a service stores credentials for the people it serves');
+ return s.id;
+ };
+ r.put('/api/v1/credentials/:subject', guard, (req, res) => {
+ try {
+ const out = credentials.put(owner(req), req.params.subject, req.body || {});
+ registry.audit(req.principal.sub, 'credential.put', 'credential', `${out.owner}:${out.subject}`, { trace: req.ov && req.ov.traceId, metadata: { provider: out.provider, base_url: out.base_url } });
+ res.set('Cache-Control', 'no-store').json(out);
+ } catch (err) { sendError(res, err, req.ov); }
+ });
+ r.get('/api/v1/credentials/:subject', guard, (req, res) => {
+ try { res.set('Cache-Control', 'no-store').json(credentials.get(owner(req), req.params.subject)); } catch (err) { sendError(res, err, req.ov); }
+ });
+ r.delete('/api/v1/credentials/:subject', guard, (req, res) => {
+ try {
+ const o = owner(req);
+ const gone = credentials.remove(o, req.params.subject);
+ if (gone) registry.audit(req.principal.sub, 'credential.delete', 'credential', `${o}:${req.params.subject}`, { trace: req.ov && req.ov.traceId });
+ res.status(gone ? 204 : 404).end();
+ } catch (err) { sendError(res, err, req.ov); }
+ });
+ return r;
+}
+
+module.exports = { createCredentials, credentialsRouter, providerKeyFor };
@@ -21,14 +21,14 @@ const { createKeyStore, createAuth } = require('./auth');
@@ -21,14 +21,14 @@ const { createKeyStore, createAuth } = require('./auth');
const schemas = require('./schemas');
const schemas = require('./schemas');
const { createApp } = require('./app');
const { createApp } = require('./app');
-async function start({ config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true } = {}) {
+async function start({ config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, listen = true, credentialFetch = null } = {}) {
config = config || load(env);
config = config || load(env);
schemas.configure(config.schemaCache);
schemas.configure(config.schemaCache);
const db = openDb(config.dbPath);
const db = openDb(config.dbPath);
const registry = createRegistry(db, { clock, env });
const registry = createRegistry(db, { clock, env });
const quotas = createQuotas(db, { clock, registry });
const quotas = createQuotas(db, { clock, registry });
const cache = createCache(db, { clock });
const cache = createCache(db, { clock });
- const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log });
+ const pool = createProviderPool({ db, registry, config, clock, fetchImpl, env, log, credentialFetch });
const fetcher = createFetcher(config);
const fetcher = createFetcher(config);
const engine = createEngine({ registry, pool, fetcher, quotas, config, log });
const engine = createEngine({ registry, pool, fetcher, quotas, config, log });
// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
// ai.run.* to OpenVibe.Events through the outbox (server/events.js); before recovery, so its failures are announced.
@@ -36,14 +36,16 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
@@ -36,14 +36,16 @@ async function start({ config, clock = { now: () => Date.now() }, fetchImpl = gl
// ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).
// ai.preferences (read) and ai.usage_summary (written): Network user modules (server/user-modules.js).
const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });
const userModules = require('./user-modules').createUserModules({ db, config, env, fetchImpl, clock, log });
userModules.ensureSchema();
userModules.ensureSchema();
- const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules });
+ // A person's own provider keys (WS-O task 2): stored by the service holding their consent, used by their runs only.
+ const credentials = require('./credentials').createCredentials({ db, config, clock });
+ const runs = createRuns({ db, registry, engine, cache, quotas, config, clock, log, userModules, credentials });
seed({ registry, quotas, config, env, db });
seed({ registry, quotas, config, env, db });
const interrupted = runs.recoverInterrupted();
const interrupted = runs.recoverInterrupted();
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
if (interrupted) log.warn(`[ai] marked ${interrupted} interrupted run(s) failed (run.interrupted); callers can retry them`);
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys });
const auth = createAuth({ config, keys });
- const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log });
+ const app = createApp({ config, db, registry, pool, quotas, cache, runs, auth, keys, env, clock, fetchImpl, log, credentials });
const keyLoaded = keys.start().catch(() => null);
const keyLoaded = keys.start().catch(() => null);
const housekeeping = setInterval(() => {
const housekeeping = setInterval(() => {
new file mode 100644
new file mode 100644
@@ -0,0 +1,51 @@
@@ -0,0 +1,51 @@
+'use strict';
+/**
+ * fetch() for a base URL a person chose (their own provider key, server/credentials.js): https only, no
+ * credentials in the URL, no IP-literal hosts, and the connection goes only to public unicast addresses, checked
+ * at connect time (fetcher.js safeLookup), so a DNS answer cannot point it inside the network. No redirects; a
+ * response is capped at 8 MB. Returns a standard Response, which is all the provider adapters use.
+ */
+const https = require('https');
+const net = require('net');
+const { safeLookup } = require('../fetcher');
+
+const MAX_BYTES = 8 * 1024 * 1024;
+
+function refuse(message) {
+ const e = new Error(message);
+ e.code = 'EADDRNOTPUBLIC';
+ return e;
+}
+
+function guardedFetch(input, init = {}) {
+ return new Promise((resolve, reject) => {
+ let url;
+ try { url = new URL(String(input)); } catch { return reject(refuse('not a valid URL')); }
+ if (url.protocol !== 'https:') return reject(refuse('only https endpoints are called with a person\'s own key'));
+ if (url.username || url.password) return reject(refuse('credentials in URLs are not allowed'));
+ if (net.isIP(url.hostname.replace(/^\[|\]$/g, ''))) return reject(refuse('IP-literal hosts are not called'));
+ if (typeof init.body !== 'string' && init.body != null) return reject(refuse('only JSON bodies are sent with a person\'s own key'));
+ const headers = { ...(init.headers || {}) };
+ if (init.body != null) headers['Content-Length'] = Buffer.byteLength(init.body);
+ const req = https.request(url, { method: init.method || 'GET', headers, lookup: safeLookup });
+ const signal = init.signal;
+ const onAbort = () => { req.destroy(); reject(Object.assign(new Error('aborted'), { name: 'AbortError' })); };
+ if (signal) { if (signal.aborted) return onAbort(); signal.addEventListener('abort', onAbort, { once: true }); }
+ req.on('error', (err) => { if (signal) signal.removeEventListener('abort', onAbort); reject(err); });
+ req.on('response', (res) => {
+ const chunks = []; let bytes = 0;
+ res.on('data', (c) => { bytes += c.length; if (bytes > MAX_BYTES) { res.destroy(); req.destroy(); reject(new Error('response too large')); } else chunks.push(c); });
+ res.on('error', reject);
+ res.on('end', () => {
+ if (signal) signal.removeEventListener('abort', onAbort);
+ const h = new Headers();
+ for (const [k, v] of Object.entries(res.headers)) if (v != null) h.set(k, Array.isArray(v) ? v.join(', ') : String(v));
+ resolve(new Response(res.statusCode === 204 ? null : Buffer.concat(chunks), { status: res.statusCode, headers: h }));
+ });
+ });
+ if (init.body != null) req.write(init.body);
+ req.end();
+ });
+}
+
+module.exports = { guardedFetch, MAX_BYTES };
@@ -29,7 +29,7 @@ function callerFault(err) {
@@ -29,7 +29,7 @@ function callerFault(err) {
return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
return Number.isInteger(st) && st >= 400 && st < 500 && ![401, 403, 404, 408, 429].includes(st);
}
}
-function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console }) {
+function createProviderPool({ db, registry, config, clock = { now: () => Date.now() }, fetchImpl = globalThis.fetch, env = process.env, log = console, credentialFetch = null }) {
const cache = new Map(); // key -> { stamp, adapter }
const cache = new Map(); // key -> { stamp, adapter }
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
const stats = {}; // key -> { calls, failures, ... } (in memory; tests + /api/ready)
@@ -132,7 +132,58 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -132,7 +132,58 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
* ctx: { runId, routeKey, routeVersion, signal, logRequest(entry), debugRaw }
* ctx: { runId, routeKey, routeVersion, signal, logRequest(entry), debugRaw }
* req: canonical request (system, messages, image, json, maxTokens, temperature, cacheKey, input, filePath, ...)
* req: canonical request (system, messages, image, json, maxTokens, temperature, cacheKey, input, filePath, ...)
*/
*/
+ /**
+ * A run with a person's own key (server/credentials.js): that provider only, no fallback, no circuit breaker
+ * shared with anyone, chat operations only; the base URL they chose is reached through guardedFetch. One
+ * request-log row per attempt under provider_key byo:<owner>:<subject>, priced by the model's list price.
+ */
+ async function executeWithCredential(cred, operation, req, ctx = {}) {
+ if (operation !== 'chat') throw new AiError(422, 'credential.unsupported', `a person's own key is used for chat only, not ${operation}`);
+ const guardedFetch = credentialFetch || require('./guarded-fetch').guardedFetch;
+ const role = ctx.role && cred.models[ctx.role] ? ctx.role : null;
+ const model = (req.image && cred.models.vision) || (role && cred.models[role]) || cred.models.chat || (cred.provider === 'anthropic' ? 'claude-haiku-4-5-20251001' : 'gpt-4o-mini');
+ const record = { key: cred.providerKey, kind: cred.provider, base_url: cred.base_url, capabilities: ['chat', 'json', 'vision'], timeout_ms: 60000 };
+ const a = cred.provider === 'anthropic' ? createAnthropicProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch }) : createOpenAiProvider(record, { apiKey: cred.apiKey, fetchImpl: guardedFetch });
+ const base = { operation, provider_key: cred.providerKey, model_key: model, route_key: ctx.routeKey || null, route_version: ctx.routeVersion || null, fallback: 0, prompt_hash: ctx.promptHash || null };
+ const timeoutMs = Math.max(1000, Math.min(req.timeoutMs || 60000, 120000));
+ let lastErr = null;
+ for (let attempt = 0; attempt <= config.providerRetries; attempt++) {
+ if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
+ const t0 = Date.now();
+ try {
+ const result = await a.chat({ ...req, model, signal: ctx.signal, timeoutMs });
+ const usage = result.usage || { input: 0, output: 0, cached: 0 };
+ const listPrice = (() => {
+ const table = config.pricing.table || {};
+ const m = String(result.model || model).toLowerCase();
+ let best = null; let len = -1;
+ for (const [k, v] of Object.entries(table)) { const key = String(k).toLowerCase(); if (key !== 'default' && m.startsWith(key) && key.length > len && v && typeof v === 'object') { best = v; len = key.length; } }
+ const b = best || table.default || {};
+ const inRate = Number(b.in) || 0;
+ return { in: inRate, out: Number(b.out) || 0, cached: b.cached != null ? Number(b.cached) : inRate * 0.1 };
+ })();
+ const input = Math.max(0, (usage.input || 0) - (usage.cached || 0));
+ const cost = (input / 1e6) * listPrice.in + ((usage.cached || 0) / 1e6) * listPrice.cached + ((usage.output || 0) / 1e6) * listPrice.out;
+ const latency = Date.now() - t0;
+ if (ctx.logRequest) {
+ ctx.logRequest({ ...base, model_key: result.model || model, status: 'ok', output_hash: sha256(result.json || result.text || ''), tokens_in: usage.input || 0, tokens_out: usage.output || 0,
+ tokens_cached: usage.cached || 0, tokens_estimated: usage.estimated ? 1 : 0, cost_usd: cost, latency_ms: latency });
+ }
+ return { result, provider: cred.providerKey, providerKind: cred.provider, model: result.model || model, fallbackUsed: false, usage, cost, latencyMs: latency, synthetic: false, tried: [], startedAt: t0 };
+ } catch (err) {
+ lastErr = err;
+ if (ctx.signal && ctx.signal.aborted) throw new AiError(409, 'run.cancelled', 'run was cancelled');
+ if (ctx.logRequest) ctx.logRequest({ ...base, status: err && err.code === 'provider.timeout' ? 'timeout' : 'error', latency_ms: Date.now() - t0, error: String(err && err.message || err).slice(0, 500) });
+ if (err && err.code === 'EADDRNOTPUBLIC') throw new AiError(422, 'credential.endpoint_refused', err.message);
+ if (attempt < config.providerRetries && retryable(err)) { await sleep(config.retryDelayMs, ctx.signal).catch(() => {}); continue; }
+ break;
+ }
+ }
+ throw new AiError(502, 'credential.provider_failed', `the person's own provider did not answer: ${String(lastErr && lastErr.message || lastErr).slice(0, 200)}`);
+ }
+
async function execute(route, operation, req, ctx = {}) {
async function execute(route, operation, req, ctx = {}) {
+ if (ctx.credential) return executeWithCredential(ctx.credential, operation, req, ctx);
const features = [operation];
const features = [operation];
if (req.image) features.push('vision');
if (req.image) features.push('vision');
if (req.json) features.push('json');
if (req.json) features.push('json');
@@ -201,7 +252,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
@@ -201,7 +252,7 @@ function createProviderPool({ db, registry, config, clock = { now: () => Date.no
throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried });
throw new AiError(503, 'provider.unavailable', 'no provider on this route could answer', { tried });
}
}
- return { adapter, health, resetHealth, execute, priceFor, costOf, stats, statFor, candidates };
+ return { adapter, health, resetHealth, execute, executeWithCredential, priceFor, costOf, stats, statFor, candidates };
}
}
module.exports = { createProviderPool };
module.exports = { createProviderPool };
@@ -58,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {
@@ -58,7 +58,7 @@ function retainedInput(input, wf, debugRaw) {
return { value, full: !redacted };
return { value, full: !redacted };
}
}
-function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null }) {
+function createRuns({ db, registry, engine, cache, quotas, config, clock = { now: () => Date.now() }, log = console, userModules = null, credentials = null }) {
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const inflight = new Map(); // run id -> { controller, promise, release, caller }
const queue = []; // run ids waiting for the worker
const queue = []; // run ids waiting for the worker
let active = 0;
let active = 0;
@@ -168,6 +168,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -168,6 +168,13 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
const routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;
const routeKey = (wf.steps[0] && wf.steps[0].route) || (tpl && tpl.default_route) || wf.default_route || null;
const route = routeKey && !String(routeKey).includes('{') ? registry.resolveRoute(routeKey) : null;
const route = routeKey && !String(routeKey).includes('{') ? registry.resolveRoute(routeKey) : null;
const options = body.options && typeof body.options === 'object' ? body.options : {};
const options = body.options && typeof body.options === 'object' ? body.options : {};
+ // A person's own key (WS-O task 2): only the service that stored it, within its daily budget; never cached.
+ const credentialSubject = body.credential && typeof body.credential === 'object' ? String(body.credential.subject || '') : null;
+ if (credentialSubject !== null) {
+ if (!credentials) throw new AiError(503, 'credentials.unavailable', 'person credentials are not configured on this service');
+ if (requester.type !== 'service') throw new AiError(403, 'capability.denied', 'only the service that stored a credential runs with it');
+ credentials.admit(requester.id, credentialSubject);
+ }
const ctx = {
const ctx = {
requesterType: requester.type, requesterId: requester.id,
requesterType: requester.type, requesterId: requester.id,
actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,
actorKey: subjectKey(body.on_behalf_of), targetKey: entityKey(body.target), attributionKey: entityKey(body.attribution), workflowKey: wf.key,
@@ -184,12 +191,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -184,12 +191,12 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
source_service: requester.type === 'service' ? requester.id : null,
source_service: requester.type === 'service' ? requester.id : null,
target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,
target: body.target ? JSON.stringify(body.target) : null, target_key: ctx.targetKey,
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
input: null, input_hash: inputHash, idempotency_key: idem, trace_id: trace, request_id: requestId, retry_of: retryOf,
- options: JSON.stringify({ cache: options.cache !== false, debug: Boolean(options.debug), input_retained: kept.full }), created_at: now,
+ options: JSON.stringify({ cache: options.cache !== false && credentialSubject === null, debug: Boolean(options.debug), input_retained: kept.full, ...(credentialSubject !== null ? { credential_subject: credentialSubject } : {}) }), created_at: now,
};
};
base.input = stableStringify(kept.value);
base.input = stableStringify(kept.value);
// Cache (scoped; never crosses requester/actor/target/attribution)
// Cache (scoped; never crosses requester/actor/target/attribution)
- const scope = options.cache === false ? null : cache.scopeFor(wf.cache_mode, ctx);
+ const scope = options.cache === false || credentialSubject !== null ? null : cache.scopeFor(wf.cache_mode, ctx);
let cacheKey = null;
let cacheKey = null;
if (scope) {
if (scope) {
cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });
cacheKey = cache.keyFor({ scope, workflowKey: wf.key, workflowVersion: wf.version, templateVersion: base.template_version, routeKey: base.route_key, routeVersion: base.route_version, inputHash });
@@ -215,7 +222,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -215,7 +222,8 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
// Queue caps, then quota: both before any provider call (a refused run is not counted).
// Queue caps, then quota: both before any provider call (a refused run is not counted).
const caller = `${requester.type}:${requester.id}`;
const caller = `${requester.type}:${requester.id}`;
admit(caller);
admit(caller);
- const reserved = quotas.reserve(ctx);
+ // A run with a person's own key spends their money, not the shared quotas (its own budget was checked above).
+ const reserved = credentialSubject !== null ? null : quotas.reserve(ctx);
db.transaction(() => {
db.transaction(() => {
db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
db.prepare(`INSERT INTO runs (id, workflow_key, workflow_version, template_key, template_version, route_key, route_version, status, requester_type, requester_id, on_behalf_of, attribution,
@@ -284,7 +292,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -284,7 +292,9 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (preferences.history === false) {
if (preferences.history === false) {
db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = json_set(options, '$.input_retained', json('false'), '$.history', json('false')) WHERE id = ?").run(id);
db.prepare("UPDATE runs SET input = 'null', cache_key = NULL, options = json_set(options, '$.input_retained', json('false'), '$.history', json('false')) WHERE id = ?").run(id);
}
}
- const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences });
+ const credential = opts.credential_subject ? credentials.forRun(row.requester_id, opts.credential_subject) : null;
+ const r = await engine.execute(wf, row, input, { signal: controller.signal, logRequest, debugRaw: Boolean(config.debugRawLog && opts.debug), inputHash: row.input_hash, preferences,
+ ...(credential ? { credential, role: input && typeof input.role === 'string' ? input.role : null } : {}) });
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
if (controller.signal.aborted) throw new AiError(409, 'run.cancelled', 'cancelled');
// The status, its citations and the ai.run.succeeded event commit together.
// The status, its citations and the ai.run.succeeded event commit together.
const done = db.transaction(() => {
const done = db.transaction(() => {
@@ -298,7 +308,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -298,7 +308,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
return d;
return d;
})();
})();
if (!done.changes) return; // cancelled while finishing
if (!done.changes) return; // cancelled while finishing
- if (r.provider) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
+ if (r.provider && reserved) quotas.account(ctx, reserved, { provider: r.provider, model: r.model, tokensIn: r.usage.input, tokensOut: r.usage.output, tokensCached: r.usage.cached, cost: r.cost });
if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
if (r.fallbackUsed) registry.audit('system', 'run.fallback', 'run', id, { trace: row.trace_id, metadata: { workflow: row.workflow_key, provider: r.provider, route: r.route && r.route.key } });
// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
// Output shaped by someone's preferences, or kept for nobody (history off), is never cached for reuse.
if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
if (scope && row.cache_key && !r.synthetic && wf.cache_mode !== 'none' && preferences.history !== false && !preferenceLines(preferences)) {
@@ -317,7 +327,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
@@ -317,7 +327,7 @@ function createRuns({ db, registry, engine, cache, quotas, config, clock = { now
if (f.changes) events.runChanged(getRow(id));
if (f.changes) events.runChanged(getRow(id));
})();
})();
const spent = db.prepare("SELECT COALESCE(SUM(tokens_in),0) ti, COALESCE(SUM(tokens_out),0) tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
const spent = db.prepare("SELECT COALESCE(SUM(tokens_in),0) ti, COALESCE(SUM(tokens_out),0) tout, COALESCE(SUM(cost_usd),0) c FROM requests WHERE run_id = ? AND status = 'ok'").get(id);
- if (spent && (spent.ti || spent.tout)) quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });
+ if (reserved && spent && (spent.ti || spent.tout)) quotas.account(ctx, reserved, { provider: 'mixed', model: '', tokensIn: spent.ti, tokensOut: spent.tout, cost: spent.c });
} finally {
} finally {
inflight.delete(id);
inflight.delete(id);
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,99 @@
@@ -0,0 +1,99 @@
+'use strict';
+// A person's own provider key (roadmap WS-O task 2; Contracts 0.73.0 ai.credential.manage):
+// - stored by a service holding ai.credential.manage, encrypted at rest and bound to its owner and subject, never
+// returned (a four-character hint only); another service neither sees nor uses it; off without AI_CREDENTIALS_KEY;
+// - a run naming it calls that provider with that key only (no fallback, never cached, the shared quota untouched),
+// logged under byo:<owner>:<subject>, and its own daily budget stops further runs (429);
+// - the guarded fetch reaches only public https endpoints.
+const assert = require('assert');
+const { boot, request, token, suite, ALL } = require('./helpers');
+const { guardedFetch } = require('../server/providers/guarded-fetch');
+
+const t = suite('credentials');
+const KEY = 'ab'.repeat(32);
+const DANA = 'usr_01JAB2C3D4E5F6G7H8J9K0MNP1';
+const API_KEY = 'sk-test-dana-0123456789WXYZ';
+const live = token('live', [...ALL, 'ai.credential.manage']);
+const tools = token('tools', [...ALL, 'ai.credential.manage']);
+const noCap = token('live', ALL);
+const calls = [];
+// The person's provider: an OpenAI-compatible endpoint, answered in-process (the guarded fetch is tested below).
+const credentialFetch = async (url, init) => {
+ calls.push({ url, auth: init.headers.Authorization || init.headers.authorization, body: JSON.parse(init.body) });
+ return new Response(JSON.stringify({ model: 'gpt-4o-mini', choices: [{ message: { role: 'assistant', content: 'hello from dana\'s key' } }], usage: { prompt_tokens: 1000, completion_tokens: 1000 } }), { status: 200, headers: { 'content-type': 'application/json' } });
+};
+let h;
+
+t.test('boot with a credentials key', async () => {
+ h = await boot({ env: { AI_CREDENTIALS_KEY: KEY, AI_PRICING_JSON: JSON.stringify({ 'gpt-4o-mini': { in: 1, out: 1 } }) }, credentialFetch });
+});
+
+t.test('store, read without the key, refuse other services', async () => {
+ const put = { provider: 'openai', base_url: 'https://llm.example.com/v1', api_key: API_KEY, models: { chat: 'gpt-4o-mini' }, budget_usd_per_day: 0.003 };
+ let r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: noCap, body: put });
+ assert.strictEqual(r.status, 403, 'needs ai.credential.manage');
+ r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { ...put, base_url: 'http://10.0.0.1/v1' } });
+ assert.strictEqual(r.status, 422, 'https only');
+ r = await request(h.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: put });
+ assert.strictEqual(r.status, 200, r.text);
+ assert.deepStrictEqual([r.body.owner, r.body.key_hint, r.body.provider], ['live', '…WXYZ', 'openai']);
+ assert.ok(!r.text.includes(API_KEY), 'the key never comes back');
+ const { validate } = require('openvibe-contracts');
+ assert.ok(validate('ai.credential@1', r.body).valid, JSON.stringify(validate('ai.credential@1', r.body).errors));
+ const row = h.db.prepare('SELECT key_enc FROM subject_credentials WHERE owner = ? AND subject = ?').get('live', DANA);
+ assert.ok(row.key_enc.startsWith('v1:') && !row.key_enc.includes(API_KEY), 'encrypted at rest');
+ r = await request(h.base, 'GET', `/api/v1/credentials/${DANA}`, { tok: tools });
+ assert.strictEqual(r.status, 404, 'another service does not see it');
+ // A row moved to another owner does not decrypt: the key is bound to owner and subject.
+ h.db.prepare("INSERT INTO subject_credentials (owner, subject, provider, base_url, key_enc, key_hint, models, created_at, updated_at) SELECT 'tools', subject, provider, base_url, key_enc, key_hint, models, created_at, updated_at FROM subject_credentials WHERE owner = 'live'").run();
+ const creds = require('../server/credentials').createCredentials({ db: h.db, config: h.config });
+ assert.throws(() => creds.forRun('tools', DANA));
+ h.db.prepare("DELETE FROM subject_credentials WHERE owner = 'tools'").run();
+});
+
+t.test('a run with the credential uses that key only, uncached, and its budget stops it', async () => {
+ const run = (tok = live) => request(h.base, 'POST', '/api/v1/runs?wait=5000', { tok, body: { workflow: 'live.viewers.line', input: { role: 'chat', user: 'say hi' }, credential: { subject: DANA } } });
+ let r = await run();
+ assert.strictEqual(r.status, 201, r.text);
+ assert.strictEqual(r.body.run.status, 'succeeded', JSON.stringify(r.body.run));
+ assert.strictEqual(r.body.run.output.text, "hello from dana's key");
+ assert.strictEqual(calls.length, 1);
+ assert.strictEqual(calls[0].url, 'https://llm.example.com/v1/chat/completions');
+ assert.strictEqual(calls[0].auth, `Bearer ${API_KEY}`);
+ const reqs = h.db.prepare('SELECT provider_key, status, cost_usd FROM requests WHERE run_id = ?').all(r.body.run.id);
+ assert.deepStrictEqual(reqs.map((x) => [x.provider_key, x.status]), [[`byo:live:${DANA}`, 'ok']]);
+ assert.ok(Math.abs(reqs[0].cost_usd - 0.002) < 1e-9, 'priced at list price');
+ r = await run();
+ assert.strictEqual(r.body.run.status, 'succeeded');
+ assert.strictEqual(calls.length, 2, 'never served from the cache');
+ r = await run();
+ assert.deepStrictEqual([r.status, r.body.code], [429, 'quota.exceeded'], 'the daily budget (0.003) is spent');
+ assert.strictEqual(calls.length, 2);
+ r = await request(h.base, 'GET', `/api/v1/credentials/${DANA}`, { tok: live });
+ assert.ok(Math.abs(r.body.spent_usd_today - 0.004) < 1e-9);
+ // Another service cannot run with it; neither can a person's token.
+ r = await run(tools);
+ assert.deepStrictEqual([r.status, r.body.code], [404, 'credential.not_found']);
+ r = await run(token('dana', ALL, { sub: DANA, actorType: 'user' }));
+ assert.ok([401, 403].includes(r.status), `a user token: ${r.status}`);
+ // Deleted: runs refuse it.
+ assert.strictEqual((await request(h.base, 'DELETE', `/api/v1/credentials/${DANA}`, { tok: live })).status, 204);
+ r = await run();
+ assert.deepStrictEqual([r.status, r.body.code], [404, 'credential.not_found']);
+});
+
+t.test('the guarded fetch reaches only public https endpoints', async () => {
+ for (const [url, why] of [['http://example.com/v1', 'https only'], ['https://127.0.0.1/v1', 'IP literal'], ['https://user:[email protected]/v1', 'credentials in URL'], ['https://localhost:9/v1', 'resolves to loopback']]) {
+ await assert.rejects(guardedFetch(url, { method: 'POST', headers: {}, body: '{}' }), (e) => e.code === 'EADDRNOTPUBLIC', why);
+ }
+});
+
+t.test('without AI_CREDENTIALS_KEY the routes answer 503', async () => {
+ const off = await boot({});
+ const r = await request(off.base, 'PUT', `/api/v1/credentials/${DANA}`, { tok: live, body: { provider: 'openai', api_key: API_KEY } });
+ assert.deepStrictEqual([r.status, r.body.code], [503, 'credentials.unavailable']);
+ await off.stop();
+ await h.stop();
+});
+
+t.run();
@@ -42,14 +42,14 @@ function tmpDir() {
@@ -42,14 +42,14 @@ function tmpDir() {
}
}
/** Boot the service. `env` goes through config.load() and is also what secrets resolve against. */
/** Boot the service. `env` goes through config.load() and is also what secrets resolve against. */
-async function boot({ env = {}, dir = tmpDir(), clock, fetchImpl } = {}) {
+async function boot({ env = {}, dir = tmpDir(), clock, fetchImpl, credentialFetch } = {}) {
const fullEnv = {
const fullEnv = {
NODE_ENV: 'test', PORT: '0', AI_DB_PATH: path.join(dir, 'ai.db'), OV_NETWORK_PUBLIC_KEY: publicKey,
NODE_ENV: 'test', PORT: '0', AI_DB_PATH: path.join(dir, 'ai.db'), OV_NETWORK_PUBLIC_KEY: publicKey,
AI_PROVIDER_RETRY_DELAY_MS: '5', AI_QUOTA_SERVICE_RPM: '0', AI_QUOTA_SERVICE_RPD: '0',
AI_PROVIDER_RETRY_DELAY_MS: '5', AI_QUOTA_SERVICE_RPM: '0', AI_QUOTA_SERVICE_RPD: '0',
...env,
...env,
};
};
const config = load(fullEnv);
const config = load(fullEnv);
- const h = await start({ config, env: fullEnv, log: silent, clock, fetchImpl });
+ const h = await start({ config, env: fullEnv, log: silent, clock, fetchImpl, credentialFetch });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
return { ...h, base, dir, env: fullEnv, stop: () => h.close() };
return { ...h, base, dir, env: fullEnv, stop: () => h.close() };
}
}