379 lines
21 KiB
JavaScript
379 lines
21 KiB
JavaScript
// P2: one broker holds this lifecycle controller after Store acquired its writer lock.
|
|
// Every management-list read/modify/write is serialized. Only Credentials owns files.
|
|
import { randomBytes, createHash } from 'node:crypto';
|
|
import { performance } from 'node:perf_hooks';
|
|
import { isDeepStrictEqual } from 'node:util';
|
|
import { BusError } from './broker.mjs';
|
|
export const REVOKE_TIMEOUT_MS = 5000;
|
|
export const REVOKE_POLL_MS = 25;
|
|
export const MODEL_STOP_TIMEOUT_MS = 5000;
|
|
const keyId = key => createHash('sha256').update(key).digest('hex').slice(0, 12);
|
|
const fail = code => { throw new BusError(code); };
|
|
// Row 333 R3 (#1831): an id that leaves the record joins the agent's earlier ids, so a daily key
|
|
// quota keeps following the agent. Ids only, never a key; at most 64, the oldest dropped first.
|
|
// The field stays absent until the first id leaves, as Credentials' format allows.
|
|
export const MAX_PREVIOUS_KEY_IDS = 64;
|
|
const withPrevious = (record, ids, retiredAt) => {
|
|
const previous = [...(record.previous ?? []), ...ids.map(id => ({ keyId: id, retiredAt }))].slice(-MAX_PREVIOUS_KEY_IDS);
|
|
const { previous: _, ...rest } = record;
|
|
return previous.length ? { ...rest, previous } : rest;
|
|
};
|
|
export class ModelKeys {
|
|
#credentials; #client; #baseUrl; #prefix; #receipt; #lock; #org; #now; #runs; #stopRuns; #grace;
|
|
#tail = Promise.resolve(); #ready = false; #failure = 'proxy-down'; #confirmed = new Map();
|
|
constructor({ credentials, client, configDir, baseUrl, receipt = () => {}, now = () => Date.now(),
|
|
runs = () => [], stopRuns = async () => {}, rotateGraceMinutes = 120, org = null, lock = null }) {
|
|
this.#credentials = credentials; this.#client = client; this.#baseUrl = baseUrl;
|
|
if ((org === null) !== (lock === null) || (org !== null && !/^[a-z0-9][a-z0-9._-]{0,63}$/.test(org))) fail('proxy-lock-invalid');
|
|
this.#org = org; this.#lock = lock;
|
|
this.#prefix = org === null ? `msk-${keyId(configDir)}-` : `mso-${keyId(configDir)}-${org}-`; this.#receipt = receipt; this.#now = now;
|
|
this.#runs = runs; this.#stopRuns = stopRuns; this.#grace = rotateGraceMinutes * 60_000;
|
|
}
|
|
#exclusive(operation) {
|
|
const result = this.#tail.then(async () => {
|
|
try { return await (this.#lock ? this.#lock.run(operation) : operation()); }
|
|
catch (error) { if (error.code === 'proxy-uncertain') fail('proxy-uncertain'); throw error; }
|
|
});
|
|
this.#tail = result.catch(() => {});
|
|
return result;
|
|
}
|
|
#uncertain() { try { this.#lock.assertCertain(); return false; } catch { return true; } }
|
|
#ownKey(key) { return key.startsWith(this.#prefix + 'key-') && /^[A-Za-z0-9_-]{43}$/.test(key.slice((this.#prefix + 'key-').length)); }
|
|
#instances() { return this.#credentials.status().filter(s => s.service === 'model').map(s => s.instance); }
|
|
#record(instance) { return this.#credentials.modelRecord(instance); }
|
|
hasRecord(instance) { return this.#record(instance) !== null; }
|
|
assertLaunchable(instance) { this.#launchable(instance); }
|
|
#sentinel(keys) {
|
|
const matches = keys.filter(k => k.startsWith(this.#prefix + 'sentinel-') && /^[A-Za-z0-9_-]{43}$/.test(k.slice((this.#prefix + 'sentinel-').length)));
|
|
const registered = this.#org === null ? null : this.#lock.registration(this.#org);
|
|
if (!matches.length || (this.#org !== null && (matches.length !== 1 || (registered && createHash('sha256').update(matches[0]).digest('hex') !== registered)))) {
|
|
this.#ready = false; this.#failure = 'proxy-open'; fail('proxy-open');
|
|
}
|
|
}
|
|
#launchable(instance) {
|
|
try { this.#lock?.assertCertain(); } catch { fail('proxy-uncertain'); }
|
|
if (this.#record(instance)?.state === 'revoking') fail('proxy-key-revoking');
|
|
if (!this.#ready) fail(this.#failure);
|
|
}
|
|
// P5's destructive queue read needs this instance's current identity, not
|
|
// an issued agent credential. A sentinel refusal stays closed until explicit
|
|
// reconcile; telemetry never retries around it or performs reconciliation.
|
|
verifyIdentity() {
|
|
return this.#exclusive(async () => {
|
|
if (!this.#ready) fail(this.#failure);
|
|
let keys;
|
|
try { keys = await this.#client.getKeys(); } catch { fail('proxy-down'); }
|
|
this.#sentinel(keys);
|
|
});
|
|
}
|
|
async #replace(before, after) {
|
|
this.#sentinel(before); this.#sentinel(after);
|
|
if (this.#lock) {
|
|
const latest = await this.#client.getKeys();
|
|
if (!isDeepStrictEqual(latest, before)) fail('proxy-client-response');
|
|
this.#lock.fence({ org: this.#org, op: 'keys.put' });
|
|
}
|
|
await this.#client.putKeys(after);
|
|
const confirmed = await this.#client.getKeys();
|
|
this.#sentinel(confirmed);
|
|
const removed = before.filter(k => !after.includes(k));
|
|
if (after.some(k => !confirmed.includes(k)) || removed.some(k => confirmed.includes(k))) fail('proxy-client-response');
|
|
this.#lock?.confirm();
|
|
}
|
|
async #pollRevoked(keys) {
|
|
const deadline = performance.now() + REVOKE_TIMEOUT_MS;
|
|
for (const key of keys) {
|
|
for (;;) {
|
|
const remaining = deadline - performance.now();
|
|
if (remaining <= 0) fail('credential-revoke-failed');
|
|
let status;
|
|
try { status = await this.#client.probeKey(key, { timeout: remaining }); }
|
|
catch { status = 0; }
|
|
if (status === 401) break;
|
|
const left = deadline - performance.now();
|
|
if (left <= 0) fail('credential-revoke-failed');
|
|
await new Promise(resolve => setTimeout(resolve, Math.min(REVOKE_POLL_MS, left)));
|
|
}
|
|
}
|
|
}
|
|
async #removeKeys(keys, confirm = () => {}) {
|
|
const before = await this.#client.getKeys();
|
|
confirm();
|
|
await this.#replace(before, before.filter(k => !keys.includes(k)));
|
|
await this.#pollRevoked(keys);
|
|
}
|
|
#captureRevocation(instance) {
|
|
const record = this.#record(instance);
|
|
if (record?.state !== 'revoking' || [record.key, ...record.retiring.map(r => r.key)].some(key => this.#org === null ? !key.startsWith(this.#prefix) : !this.#ownKey(key))) fail('credential-revoke-failed');
|
|
return { record, snapshot: this.#credentials.captureModelRevocation(instance) };
|
|
}
|
|
#confirmRevocation(instance, proof) {
|
|
if (!isDeepStrictEqual(this.#record(instance), proof.record)) fail('credential-revoke-failed');
|
|
// A prior tmp removal may leave only final at retry/restart. Neither its
|
|
// absence nor a revoking label proves exit. Fresh views grant NO stop authority.
|
|
if (this.#runs(instance).length) fail('proxy-stop-failed');
|
|
}
|
|
async reconcile() {
|
|
return this.#exclusive(async () => {
|
|
this.#ready = false;
|
|
try {
|
|
let before = await this.#client.getKeys();
|
|
const sentinels = () => before.filter(k => k.startsWith(this.#prefix + 'sentinel-') && /^[A-Za-z0-9_-]{43}$/.test(k.slice((this.#prefix + 'sentinel-').length)));
|
|
const registered = this.#org === null ? null : this.#lock.registration(this.#org);
|
|
if (this.#org !== null && (sentinels().length > 1 || (registered && (sentinels().length !== 1 || createHash('sha256').update(sentinels()[0]).digest('hex') !== registered)))) fail('proxy-open');
|
|
if (this.#org !== null && sentinels().length === 0) {
|
|
const sentinel = this.#prefix + 'sentinel-' + randomBytes(32).toString('base64url');
|
|
const latest = await this.#client.getKeys();
|
|
if (!isDeepStrictEqual(latest, before)) fail('proxy-client-response');
|
|
this.#lock.fence({ org: this.#org, op: 'sentinel.register' });
|
|
await this.#client.putKeys([...before, sentinel]);
|
|
const confirmed = await this.#client.getKeys();
|
|
if (!confirmed.includes(sentinel) || before.some(k => !confirmed.includes(k))) fail('proxy-client-response');
|
|
this.#lock.register(this.#org, sentinel);
|
|
this.#lock.confirm();
|
|
this.#receipt({ action: 'proxy.sentinel.registered', org: this.#org });
|
|
before = confirmed;
|
|
}
|
|
this.#sentinel(before);
|
|
if (this.#org !== null && !registered) this.#lock.register(this.#org, sentinels()[0]);
|
|
const records = this.#instances().map(instance => ({ instance, record: this.#record(instance) })).filter(e => e.record);
|
|
const retained = records.filter(e => e.record.state !== 'revoking').flatMap(e => [e.record.key, ...e.record.retiring.map(r => r.key)]);
|
|
if (this.#org !== null && retained.some(key => !this.#ownKey(key))) fail('proxy-open');
|
|
const revoking = records.filter(e => e.record.state === 'revoking');
|
|
const proofs = new Map(revoking.map(({ instance }) => { this.#confirmed.delete(instance); return [instance, this.#captureRevocation(instance)]; }));
|
|
const confirm = () => { for (const [instance, proof] of proofs) this.#confirmRevocation(instance, proof); };
|
|
confirm();
|
|
const intended = revoking.flatMap(e => [e.record.key, ...e.record.retiring.map(r => r.key)]);
|
|
const removed = before.filter(k => (this.#org === null ? k.startsWith(this.#prefix) && !k.startsWith(this.#prefix + 'sentinel-') : this.#ownKey(k)) && !retained.includes(k));
|
|
const after = [...new Set([...before.filter(k => !removed.includes(k) && !intended.includes(k)), ...retained])];
|
|
const added = after.filter(k => !before.includes(k));
|
|
if (added.length || removed.length || intended.length) await this.#replace(before, after);
|
|
await this.#pollRevoked([...new Set([...removed, ...intended])]);
|
|
confirm();
|
|
for (const { instance } of revoking) {
|
|
const proof = proofs.get(instance);
|
|
this.#credentials.prepareModelCompletion(instance, proof.snapshot);
|
|
this.#confirmed.set(instance, proof);
|
|
this.completeRevoke(instance);
|
|
}
|
|
this.#ready = true;
|
|
const receipt = { action: 'proxy.reconcile', addedKeyIds: added.map(keyId), removedKeyIds: [...new Set([...removed, ...intended])].map(keyId) };
|
|
this.#receipt(receipt); return receipt;
|
|
} catch (error) {
|
|
this.#failure = error.code === 'proxy-uncertain' || (this.#lock && this.#uncertain()) ? 'proxy-uncertain' : error.code === 'proxy-open' ? 'proxy-open' : 'proxy-down';
|
|
this.#receipt({ action: 'proxy.reconcile', code: this.#failure });
|
|
fail(this.#failure);
|
|
}
|
|
});
|
|
}
|
|
async #issue(instance, rotate = false, currentKeys = null) {
|
|
this.#launchable(instance); this.#credentials.registerModel(instance); this.#launchable(instance);
|
|
const old = this.#record(instance);
|
|
let before;
|
|
try { before = currentKeys ?? await this.#client.getKeys(); this.#sentinel(before); }
|
|
catch (error) { if (error.code === 'proxy-open') throw error; fail('proxy-key-issue-failed'); }
|
|
if (this.#org !== null && old && !this.#ownKey(old.key)) fail('proxy-open');
|
|
if (old && !rotate && before.includes(old.key)) return { baseUrl: this.#baseUrl, key: old.key };
|
|
const key = this.#prefix + (this.#org === null ? '' : 'key-') + randomBytes(32).toString('base64url');
|
|
const now = this.#now(), issuedAt = new Date(now).toISOString();
|
|
const date = new Date(now + 30 * 86400_000).toISOString().slice(0, 10);
|
|
const keep = rotate && before.includes(old?.key);
|
|
const retiring = old ? [...old.retiring, ...(keep ? [{ key: old.key, keyId: old.keyId, issuedAt: old.issuedAt, retireAt: new Date(now + this.#grace).toISOString() }] : [])] : [];
|
|
// A re-issue, or a rotation the proxy already lost the key for, drops the old key at once.
|
|
const record = withPrevious({ key, keyId: keyId(key), issuedAt, rotateBy: date, state: 'valid', retiring, previous: old?.previous },
|
|
old && !keep ? [old.keyId] : [], issuedAt);
|
|
try {
|
|
await this.#replace(before, [...new Set([...before, key])]);
|
|
this.#credentials.writeModel(instance, record);
|
|
this.#receipt({ action: rotate ? 'proxy.key.rotated' : 'proxy.key.issued', keyIds: [keyId(key)], instance });
|
|
return { baseUrl: this.#baseUrl, key };
|
|
} catch (error) {
|
|
if (error.code === 'proxy-open' || error.code === 'proxy-uncertain') throw error;
|
|
if (this.#lock && this.#uncertain()) fail('proxy-uncertain');
|
|
fail('proxy-key-issue-failed');
|
|
}
|
|
}
|
|
ensure(instance) { return this.#exclusive(() => this.#issue(instance)); }
|
|
rotate(instance) { return this.#exclusive(() => this.#issue(instance, true)); }
|
|
// Host-only callbacks are the existing synchronous organization path.
|
|
hire(instance, check, publish) {
|
|
return this.#exclusive(async () => {
|
|
await check(); // Fresh validation and captured revision AFTER controller queue.
|
|
const existed = this.hasRecord(instance);
|
|
try {
|
|
await this.#issue(instance);
|
|
await check(); // Recheck captured revision AFTER HTTP, before publication.
|
|
return await publish();
|
|
} catch (error) {
|
|
// A newly issued generation has not escaped to queued admission yet.
|
|
if (!existed && this.hasRecord(instance)) {
|
|
try { await this.#prepareRevoke(instance); this.completeRevoke(instance); } catch {}
|
|
}
|
|
throw error;
|
|
}
|
|
});
|
|
}
|
|
async preflight(instance) {
|
|
// The revoking check must run before even an issue/preflight request.
|
|
this.#launchable(instance);
|
|
const proxy = await this.#exclusive(async () => {
|
|
this.#launchable(instance);
|
|
let keys;
|
|
try { keys = await this.#client.getKeys(); } catch { fail('proxy-down'); }
|
|
this.#sentinel(keys);
|
|
return this.#issue(instance, false, keys);
|
|
});
|
|
let status;
|
|
try { status = await this.#client.probeKey(proxy.key); }
|
|
catch { fail('proxy-down'); }
|
|
if (status < 200 || status >= 300) fail('proxy-key-refused');
|
|
return proxy;
|
|
}
|
|
async #prepareRevoke(instance) {
|
|
const record = this.#record(instance);
|
|
if (!record) { if (this.#confirmed.has(instance)) fail('credential-revoke-failed'); return; }
|
|
this.#confirmed.delete(instance);
|
|
if (record.state !== 'revoking') this.#credentials.writeModel(instance, { ...record, state: 'revoking', revokingAt: new Date(this.#now()).toISOString() });
|
|
try {
|
|
const proof = this.#captureRevocation(instance), confirm = () => this.#confirmRevocation(instance, proof);
|
|
confirm();
|
|
await this.#removeKeys([record.key, ...record.retiring.map(r => r.key)], confirm);
|
|
confirm();
|
|
this.#credentials.prepareModelCompletion(instance, proof.snapshot);
|
|
this.#confirmed.set(instance, proof);
|
|
this.#receipt({ action: 'proxy.key.revoked', keyIds: [record.keyId, ...record.retiring.map(r => r.keyId)], remoteState: 'revoked', localState: 'kept' });
|
|
} catch (error) { if (error.code === 'proxy-uncertain' || (this.#lock && this.#uncertain())) fail('proxy-uncertain'); fail('credential-revoke-failed'); }
|
|
}
|
|
prepareRevoke(instance, beforeRevoke = () => {}) { return this.#exclusive(() => { beforeRevoke(); return this.#prepareRevoke(instance); }); }
|
|
completeRevoke(instance) {
|
|
try { this.#lock?.assertCertain(); } catch { fail('proxy-uncertain'); }
|
|
try {
|
|
const record = this.#record(instance), proof = this.#confirmed.get(instance);
|
|
if (!record && !proof) return; // Ordinary no-record/no-supplied-proof idempotence.
|
|
if (!record || !proof || record.state !== 'revoking') fail('credential-revoke-failed');
|
|
this.#confirmRevocation(instance, proof);
|
|
this.#credentials.finishModelRevocation(instance, proof.snapshot);
|
|
this.#confirmed.delete(instance);
|
|
} catch { fail('credential-revoke-failed'); }
|
|
}
|
|
revisionConflict(instance) {
|
|
const record = this.#record(instance);
|
|
this.#receipt({ action: 'proxy.key.revoke-conflict', keyIds: record ? [record.keyId] : [], remoteState: 'revoked', localState: 'kept' });
|
|
fail('proxy-revoke-revision-conflict');
|
|
}
|
|
sweep() {
|
|
return this.#exclusive(async () => {
|
|
let firstError;
|
|
for (const instance of this.#instances()) {
|
|
try {
|
|
let record = this.#record(instance);
|
|
if (!record || record.state === 'revoking') continue;
|
|
if (this.#now() >= Date.parse(record.rotateBy + 'T00:00:00Z')) {
|
|
await this.#issue(instance, true); record = this.#record(instance);
|
|
}
|
|
for (const old of record.retiring) {
|
|
const runs = this.#runs(instance).filter(r => r.keyId === old.keyId);
|
|
if (runs.length && this.#now() < Date.parse(old.retireAt)) continue;
|
|
if (runs.length) {
|
|
try { await this.#stopRuns(runs); } catch { fail('proxy-stop-failed'); }
|
|
}
|
|
if (this.#runs(instance).some(r => r.keyId === old.keyId)) fail('proxy-stop-failed');
|
|
const confirm = () => {
|
|
if (this.#runs(instance).some(r => r.keyId === old.keyId)) fail('proxy-stop-failed');
|
|
};
|
|
await this.#removeKeys([old.key], confirm);
|
|
confirm();
|
|
const current = this.#record(instance);
|
|
this.#credentials.writeModel(instance, withPrevious({ ...current, retiring: current.retiring.filter(r => r.keyId !== old.keyId) },
|
|
[old.keyId], new Date(this.#now()).toISOString()));
|
|
this.#receipt({ action: 'proxy.key.retired', keyIds: [old.keyId] });
|
|
}
|
|
} catch (error) {
|
|
firstError ??= error;
|
|
this.#receipt({ action: 'proxy.sweep', instance, outcome: 'refused', code: error.code ?? 'proxy-stop-failed' });
|
|
}
|
|
}
|
|
if (firstError) throw firstError;
|
|
});
|
|
}
|
|
}
|
|
|
|
// No new processing-control stop reason/schema: stopSession owns bounded SIGTERM
|
|
// and start-time verification. This layer additionally checks agent/key ownership.
|
|
export async function stopModelRuns({ runs, sessions, stop, receipt, reason = 'proxy-rotation', timeoutMs = MODEL_STOP_TIMEOUT_MS }) {
|
|
for (const run of runs) {
|
|
let timer;
|
|
const ids = typeof run.keyId === 'string' && /^[a-f0-9]{12}$/.test(run.keyId) ? [run.keyId] : [];
|
|
try {
|
|
if (!ids.length) fail('proxy-stop-failed');
|
|
const entry = sessions().find(s => s.business === run.business && s.run === run.run);
|
|
if (!entry || entry.instance !== run.role || entry.modelKeyId !== run.keyId
|
|
|| entry.pid !== run.pid || entry.startTime !== run.startTime) fail('proxy-stop-failed');
|
|
if (entry.live) {
|
|
const result = await Promise.race([
|
|
stop(run.run),
|
|
new Promise((_, reject) => { timer = setTimeout(() => reject(new BusError('proxy-stop-failed')), timeoutMs); }),
|
|
]);
|
|
if (!result?.stopped || sessions().some(s => s.business === run.business && s.run === run.run && s.live)) fail('proxy-stop-failed');
|
|
}
|
|
receipt({ action: reason, keyIds: ids, business: run.business, agent: run.role, run: run.run, outcome: 'stopped' });
|
|
} catch {
|
|
receipt({ action: reason, code: 'proxy-stop-failed', keyIds: ids, run: run.run });
|
|
fail('proxy-stop-failed');
|
|
} finally { clearTimeout(timer); }
|
|
}
|
|
}
|
|
|
|
// Option C: HTTP is outside the organization lock. This adapter preserves write's
|
|
// synchronous contract and passes the captured revision into its existing lock check.
|
|
// Neither organization.mjs nor its synchronous settings callers needs modification.
|
|
export function modelOrganization({ modelKeys, writer }) {
|
|
const pending = new Map();
|
|
const adapter = {
|
|
...writer,
|
|
write(business, document, options) {
|
|
const prepare = pending.get(business);
|
|
return writer.write(business, document, prepare ? { ...options, revision: prepare.revision } : options);
|
|
},
|
|
};
|
|
const revoke = original => (business, agent, how, services) => {
|
|
const prepare = pending.get(business), instance = business + '/' + agent;
|
|
if (!prepare || prepare.instance !== instance) fail('credential-revoke-failed');
|
|
// For an already-delisted repeat there is no write callback/lock, but the
|
|
// revision still must match before completing the model credential cleanup.
|
|
if (writer.read(business).revision !== prepare.revision) modelKeys.revisionConflict(instance);
|
|
modelKeys.completeRevoke(instance);
|
|
return original(business, agent, how, services);
|
|
};
|
|
const wrap = invoke => (business, verb, args, actor, context) => {
|
|
const isRevoke = verb === 'instance.fire' || (verb === 'instances.set' && args.op === 'delete');
|
|
if (!isRevoke) return invoke(business, verb, args, actor, context);
|
|
return (async () => {
|
|
if (pending.has(business)) fail('organization-busy');
|
|
const agent = args.name ?? args.instance, instance = business + '/' + agent;
|
|
if (verb === 'instance.fire') invoke(business, 'instance.fireCheck', { name: agent }, actor, context);
|
|
if (verb === 'instances.set') invoke(business, 'instances.check', { instance: agent }, actor, context);
|
|
const { revision } = writer.read(business);
|
|
if (args.revision !== undefined && args.revision !== revision) fail('revision-conflict');
|
|
const prepare = { instance, revision, remoteRevoked: false };
|
|
pending.set(business, prepare);
|
|
const existed = modelKeys.hasRecord(instance);
|
|
try {
|
|
if (isRevoke) { await modelKeys.prepareRevoke(instance, () => context?.beforeWrite?.()); prepare.remoteRevoked = existed; }
|
|
return invoke(business, verb, args, actor, context);
|
|
} catch (error) {
|
|
if (isRevoke && prepare.remoteRevoked && (error.code === 'revision-conflict' || error.code === 'revision-conflict-revoked')) modelKeys.revisionConflict(instance);
|
|
throw error;
|
|
} finally { pending.delete(business); }
|
|
})();
|
|
};
|
|
const hire = async (business, agent, check, publish) => {
|
|
if (pending.has(business)) fail('organization-busy');
|
|
const prepare = { instance: `${business}/${agent}`, revision: writer.read(business).revision, remoteRevoked: false };
|
|
pending.set(business, prepare);
|
|
try { return await modelKeys.hire(prepare.instance, check, publish); }
|
|
finally { pending.delete(business); }
|
|
};
|
|
return { writer: adapter, revoke, wrap, hire };
|
|
}
|