181 lines
19 KiB
JavaScript
181 lines
19 KiB
JavaScript
// OPERATOR/CLI ONLY. Runtime host must never import this private-key module.
|
|
import pg from 'pg'
|
|
import {createPrivateKey,createPublicKey,sign} from 'node:crypto'
|
|
import {open,lstat,realpath,unlink} from 'node:fs/promises'
|
|
import {writeFile,withJournalLease} from './operator-files.mjs'
|
|
import path from 'node:path'
|
|
import {captureDatabaseSourceObservation,assertDatabaseSourceObservationCurrent} from '../../src/db/database-source-observation.js'
|
|
import {MIGRATION_MANIFEST,migrationManifestDigest} from '../../src/db/migration-manifest.js'
|
|
|
|
|
|
import {ownedContext,sameOwned,sourceCoordinates,protectedStructure,readCurrent,HISTORY,CURRENT} from './context.mjs'
|
|
import {bytes,canonical,sha,nextEpoch,POLICY_DOMAIN,HEAD_DOMAIN,KEY,liveValidity,policySchema,headSchema,publicKey} from '../../src/db/application-writer-profile-format.js'
|
|
export function createConnectedPublisher(installation,collectFixedCatalogue,sourceVerifier,approval){
|
|
const PUBLISHER=installation.roleBindings.publisher.name
|
|
class ProfilePublicationError extends Error{
|
|
automaticReplay=false
|
|
constructor(outcome,phase,proposal,cleanupUnconfirmed,cause){super('Protected policy publication requires exact inspection',{cause});Object.assign(this,{outcome,phase,proposal,cleanupUnconfirmed})}
|
|
}
|
|
async function privateKey(filename){
|
|
const resolved=await realpath(filename),meta=await lstat(filename)
|
|
if(resolved!==path.resolve(filename)||!meta.isFile()||meta.isSymbolicLink()||meta.size<1||meta.size>8192)throw Error('Explicit bounded operator-only key file required')
|
|
const handle=await open(resolved,'r');let data
|
|
try{const stat=await handle.stat();if(stat.dev!==meta.dev||stat.ino!==meta.ino||stat.size!==meta.size)throw Error('Operator key changed');data=await handle.readFile();if(data.length!==stat.size)throw Error('Operator key size changed')}finally{await handle.close()}
|
|
let key;try{key=createPrivateKey(data)}finally{data.fill(0)}
|
|
if(key.asymmetricKeyType!=='ed25519'||sha(createPublicKey(key).export({format:'der',type:'spki'}))!==installation.publicKeySpkiSha256)throw Error('Private key does not match fixed installation')
|
|
publicKey(installation);return key
|
|
}
|
|
// One retained actual Pool client; no injected SQL facade, SET ROLE, new pool,
|
|
// detached application callbacks or cleanup/COMMIT races.
|
|
async function session(pool,deadlineAt,work){
|
|
if(!(pool instanceof pg.Pool)||!Number.isSafeInteger(deadlineAt)||deadlineAt<=Date.now()||deadlineAt-Date.now()>120000)throw Error('Bounded genuine publisher pool/lifetime required')
|
|
let client,retired=false,released=false,locked=false,begun=false,commitSent=false,committed=false,confirmedValue,proposal,phase='checkout',stopReject,cleanupUnconfirmed=false
|
|
const stopped=new Promise((_,reject)=>{stopReject=reject});void stopped.catch(()=>{})
|
|
const release=bad=>{if(!client||released)return;released=true;client.off('error',fault);client.off('end',fault);try{client.release(bad)}catch{cleanupUnconfirmed=true}}
|
|
const fault=()=>{if(retired)return;retired=true;cleanupUnconfirmed=true;release(true);stopReject(Error('Publisher connection/lifetime ended'))}
|
|
const timer=setTimeout(fault,Math.max(1,deadlineAt-Date.now())),bounded=p=>Promise.race([p,stopped])
|
|
const query=async(text,values)=>{if(retired||Date.now()>=deadlineAt)throw Error('Publisher lifetime closed');return bounded(client.query(text,values))}
|
|
const state={query,phase:value=>{phase=value},proposal:value=>{proposal=Object.freeze(value)},begun:value=>{begun=value},commit:async value=>{
|
|
confirmedValue=Object.freeze({...value})
|
|
commitSent=true
|
|
const response=await bounded(client.query('COMMIT').then(result=>{if(result.command==='COMMIT'){committed=true;begun=false}return result}))
|
|
if(response.command!=='COMMIT')throw Error('Publisher COMMIT acknowledgement absent')
|
|
}}
|
|
let value,primary,failed=false
|
|
try{
|
|
const pending=pool.connect().then(value=>{client=value;if(retired){release(true);return value}client.on('error',fault);client.on('end',fault);return value});void pending.catch(()=>{})
|
|
await bounded(pending);if(retired)throw Error('Late publisher checkout')
|
|
await query("SET statement_timeout='10s'");await query("SET lock_timeout='5s'")
|
|
phase='exclusive-admission';await query('SELECT pg_catalog.pg_advisory_lock($1::bigint)',[KEY]);locked=true
|
|
value=await work(state)
|
|
}catch(error){failed=true;primary=error}
|
|
finally{
|
|
if(!retired&&client){
|
|
if(begun&&!commitSent)try{const r=await query('ROLLBACK');if(r.command!=='ROLLBACK')throw Error('Publisher rollback unconfirmed');begun=false}catch{fault()}
|
|
if(commitSent&&!committed){cleanupUnconfirmed=true;retired=true;release(true)}
|
|
if(!retired&&locked)try{const r=await query('SELECT pg_catalog.pg_advisory_unlock($1::bigint) AS released',[KEY]);if(r.rows.length!==1||r.rows[0].released!==true)throw Error('Publisher unlock unconfirmed');locked=false}catch{fault()}
|
|
if(!retired)try{await query('RESET lock_timeout');await query('RESET statement_timeout')}catch{fault()}
|
|
}
|
|
clearTimeout(timer);release(retired||cleanupUnconfirmed||begun||locked||committed&&failed)
|
|
}
|
|
if(committed)return Object.freeze({...confirmedValue,...value,outcome:'committed',proposal,cleanupUnconfirmed:cleanupUnconfirmed||retired||failed,automaticReplay:false})
|
|
if(failed||cleanupUnconfirmed||retired)throw new ProfilePublicationError(commitSent||begun?'unknown':'not-committed',phase,proposal,cleanupUnconfirmed||retired,primary)
|
|
return value
|
|
}
|
|
async function publishCurrentProfile(pool,{privateKeyFile,expectedHeadSha256,journalDirectory,validityMs=86400000,deadlineAt=Date.now()+120000}){
|
|
if(!Number.isSafeInteger(validityMs)||validityMs<1||validityMs>86400000||expectedHeadSha256!==null&&!(/^[a-f0-9]{64}$/).test(expectedHeadSha256))throw Error('Explicit previous head and bounded validity required')
|
|
const key=await privateKey(privateKeyFile),directory=await realpath(journalDirectory),pendingFile=path.join(directory,'pending-publication.json')
|
|
try{await lstat(pendingFile);throw Error('Pending publication must be reconciled before signing')}catch(error){if(error.code!=='ENOENT')throw error}
|
|
const result=await session(pool,deadlineAt,async s=>{
|
|
s.phase('publisher-read');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
|
|
const first=await ownedContext(s,installation,PUBLISHER,'publisher-read')
|
|
const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
|
|
const observation=await captureDatabaseSourceObservation(facade,options)
|
|
await sourceVerifier.verify(facade,observation,options)
|
|
if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned source mismatch')
|
|
const security=await protectedStructure(s,installation),catalogue=await collectFixedCatalogue(s)
|
|
const markers=(await s.query('SELECT id FROM public.schema_migrations ORDER BY id LIMIT 45')).rows.map(row=>row.id)
|
|
if(canonical(markers)!==canonical(MIGRATION_MANIFEST.map(pin=>pin.file.slice(0,-4))))throw Error('Exact current migration marker metadata required')
|
|
const old=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
|
|
if((old?.headSha256??null)!==expectedHeadSha256)throw Error('Expected protected current head mismatch')
|
|
const catalogSha256=sha(canonical({base:catalogue.catalogSha256,security:security.sha256}))
|
|
if(approval.expiresAt<=Date.now()||approval.manifestSha256!==migrationManifestDigest()||approval.definitionSha256!==catalogue.definitionSha256||approval.catalogSha256!==catalogSha256)throw Error('Independently approved policy differs from actual source44 catalogue')
|
|
if(old&&(canonical(old.policy.source)!==canonical(sourceCoordinates(first))||canonical(old.policy.roles)!==canonical(security.roles)||old.policy.manifestSha256!==migrationManifestDigest()||old.policy.definitionSha256!==catalogue.definitionSha256||old.policy.catalogSha256!==catalogSha256))throw Error('Renewal cannot adopt a changed source/subject/policy')
|
|
await sourceVerifier.verify(facade,observation,options);await assertDatabaseSourceObservationCurrent(facade,observation,options);sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
|
|
await s.query('COMMIT');s.begun(false)
|
|
const number=old?nextEpoch(old.epoch):'1',now=Date.now()
|
|
const body=policySchema.parse({format:'application-writer-policy-v1',installationId:installation.installationId,epoch:number,issuedAt:now,notBefore:now,notAfter:now+validityMs,source:sourceCoordinates(first),sourceObservationSha256:sha(canonical(observation)),roles:security.roles,manifestSha256:migrationManifestDigest(),definitionSha256:catalogue.definitionSha256,catalogSha256,futureLeaves:'fixed12-native-policy-v1'})
|
|
const envelope=(domain,body)=>bytes({body,signature:sign(null,Buffer.from(domain+canonical(body)),key).toString('base64url')})
|
|
const policyBytes=envelope(POLICY_DOMAIN,body),policySha256=sha(policyBytes)
|
|
const head=headSchema.parse({format:'application-writer-head-v1',installationId:installation.installationId,epoch:number,policySha256,previousHeadSha256:old?.headSha256??null})
|
|
const headBytes=envelope(HEAD_DOMAIN,head),headSha256=sha(headBytes)
|
|
const proposal={installationId:installation.installationId,epoch:number,headSha256,previousHeadSha256:head.previousHeadSha256,policySha256,publisherPid:first.pid}
|
|
s.proposal(proposal)
|
|
// Public, bounded, exclusively created journal; neither key nor SQL data.
|
|
await writeFile(path.join(directory,'proposal-'+headSha256+'.json'),bytes(proposal),{flag:'wx',mode:0o600})
|
|
await writeFile(pendingFile,bytes(proposal),{flag:'wx',mode:0o600})
|
|
s.phase('publisher-write');await s.query('BEGIN ISOLATION LEVEL READ COMMITTED READ WRITE');s.begun(true)
|
|
await s.query('SELECT pg_catalog.pg_current_xact_id()');await s.query('SELECT pg_catalog.pg_advisory_xact_lock_shared($1::bigint)',[KEY])
|
|
const owner=await ownedContext(s,installation,PUBLISHER,'publisher-write')
|
|
if(owner.pid!==first.pid||canonical(sourceCoordinates(owner))!==canonical(sourceCoordinates(first)))throw Error('Retained publisher source/backend changed')
|
|
await s.query(`SELECT id FROM mmapp_control.${CURRENT} WHERE id=1 FOR UPDATE`)
|
|
const current=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
|
|
if((current?.headSha256??null)!==expectedHeadSha256)throw Error('CAS predecessor changed')
|
|
const againSecurity=await protectedStructure(s,installation),againCatalogue=await collectFixedCatalogue(s)
|
|
if(sha(canonical({base:againCatalogue.catalogSha256,security:againSecurity.sha256}))!==catalogSha256)throw Error('Policy drift during retained exclusive publication')
|
|
if(Date.now()>=body.notAfter||Date.now()>=approval.expiresAt)throw Error('Policy or approval expired before publication dispatch')
|
|
await s.query(`INSERT INTO mmapp_control.${HISTORY}(installation_id,epoch,head_sha256,previous_installation_id,previous_epoch,previous_head_sha256,policy_sha256,policy_bytes,head_bytes) VALUES($1::uuid,$2::bigint,$3,$4::uuid,$5::bigint,$6,$7,$8::bytea,$9::bytea)`,[installation.installationId,number,headSha256,old?installation.installationId:null,old?.epoch??null,old?.headSha256??null,policySha256,policyBytes,headBytes])
|
|
const changed=old?await s.query(`UPDATE mmapp_control.${CURRENT} SET epoch=$1::bigint,head_sha256=$2 WHERE id=1 AND installation_id=$3::uuid AND epoch=$4::bigint AND head_sha256=$5 RETURNING id`,[number,headSha256,installation.installationId,old.epoch,old.headSha256]):await s.query(`INSERT INTO mmapp_control.${CURRENT}(id,installation_id,epoch,head_sha256) VALUES(1,$1::uuid,$2::bigint,$3) RETURNING id`,[installation.installationId,number,headSha256])
|
|
if(changed.rows.length!==1)throw Error('Exactly one protected current CAS required')
|
|
sameOwned(owner,await ownedContext(s,installation,PUBLISHER,'publisher-write'));if(Date.now()>=body.notAfter||Date.now()>=approval.expiresAt)throw Error('Policy or approval expired before publication terminal');s.phase('commit');const result={epoch:number,headSha256,notAfter:body.notAfter};await s.commit(result)
|
|
const terminal=await ownedContext(s,installation,PUBLISHER,'publisher-session')
|
|
if(terminal.pid!==first.pid||canonical(sourceCoordinates(terminal))!==canonical(sourceCoordinates(first)))throw Error('Publisher session continuity unconfirmed after actual ACK')
|
|
return result
|
|
})
|
|
// A journal failure cannot turn a confirmed publication into retryable failure.
|
|
if(result.outcome==='committed'&&!result.cleanupUnconfirmed){try{const cursor=await boundedPending(pendingFile);if(cursor.headSha256!==result.headSha256)throw Error('Pending cursor differs');await unlink(pendingFile)}catch{return Object.freeze({...result,journalUnconfirmed:true})}}
|
|
return result
|
|
}
|
|
async function inspectPublication(pool,proposal,{deadlineAt=Date.now()+120000}={}){
|
|
if(!proposal||proposal.installationId!==installation.installationId||!(/^[a-f0-9]{64}$/).test(proposal.headSha256)||proposal.previousHeadSha256!==null&&!(/^[a-f0-9]{64}$/).test(proposal.previousHeadSha256))throw Error('Bounded public proposal required')
|
|
return session(pool,deadlineAt,async s=>{
|
|
s.phase('publication-inspection');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
|
|
const first=await ownedContext(s,installation,PUBLISHER,'publisher-read');await protectedStructure(s,installation)
|
|
if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned inspection source mismatch')
|
|
const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
|
|
const observation=await captureDatabaseSourceObservation(facade,options);await sourceVerifier.verify(facade,observation,options)
|
|
const current=await readCurrent(s,installation,{allowEmpty:true,validNow:false})
|
|
const proposed=(await s.query(`SELECT epoch::text,head_sha256 FROM mmapp_control.${HISTORY} WHERE installation_id=$1::uuid AND epoch=$2::bigint`,[installation.installationId,proposal.epoch])).rows
|
|
await sourceVerifier.verify(facade,observation,options);sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
|
|
const response=await s.query('ROLLBACK');if(response.command!=='ROLLBACK')throw Error('Inspection terminal unconfirmed');s.begun(false)
|
|
if(current?.headSha256===proposal.headSha256&&proposed.length===1&&proposed[0].head_sha256===proposal.headSha256)return {state:'confirmed',headSha256:current.headSha256}
|
|
if((current?.headSha256??null)===proposal.previousHeadSha256&&proposed.length===0)return {state:'not-committed',headSha256:current?.headSha256??null}
|
|
throw Error('Publication state requires operator inspection; no retry')
|
|
})
|
|
}
|
|
// Read-only diagnostics on the actual protected publisher session. This does
|
|
// not admit application/operator writes, bootstrap, restore, or READY.
|
|
async function preflightCurrentProfile(pool,{deadlineAt=Date.now()+120000}={}){
|
|
return session(pool,deadlineAt,async s=>{
|
|
s.phase('profile-preflight');await s.query('BEGIN ISOLATION LEVEL REPEATABLE READ READ ONLY');s.begun(true)
|
|
const first=await ownedContext(s,installation,PUBLISHER,'publisher-read')
|
|
if(canonical(sourceCoordinates(first))!==canonical(installation.expectedSource))throw Error('Pinned preflight source mismatch')
|
|
const facade=Object.freeze({query:(sql,values)=>s.query(sql,values)}),options={deadlineAt}
|
|
const observation=await captureDatabaseSourceObservation(facade,options),initialSource=await sourceVerifier.verify(facade,observation,options)
|
|
const security=await protectedStructure(s,installation),catalogue=await collectFixedCatalogue(s)
|
|
const current=await readCurrent(s,installation)
|
|
if(!current)throw Error('Current signed policy required for preflight')
|
|
const catalogSha256=sha(canonical({base:catalogue.catalogSha256,security:security.sha256})),manifestSha256=migrationManifestDigest()
|
|
if(approval.expiresAt<=Date.now()||approval.manifestSha256!==manifestSha256||approval.definitionSha256!==catalogue.definitionSha256||approval.catalogSha256!==catalogSha256)throw Error('Independent preflight approval differs from actual source44 catalogue')
|
|
if(canonical(current.policy.source)!==canonical(sourceCoordinates(first))||canonical(current.policy.roles)!==canonical(security.roles)||current.policy.manifestSha256!==manifestSha256||current.policy.definitionSha256!==catalogue.definitionSha256||current.policy.catalogSha256!==catalogSha256)throw Error('Current signed policy differs from actual source/roles/catalogue')
|
|
const terminalSource=await sourceVerifier.verify(facade,observation,options);await assertDatabaseSourceObservationCurrent(facade,observation,options)
|
|
sameOwned(first,await ownedContext(s,installation,PUBLISHER,'publisher-read'))
|
|
// Recheck the live interval after the potentially lengthy catalogue walk.
|
|
const clock=(await s.query('SELECT (EXTRACT(EPOCH FROM pg_catalog.clock_timestamp())*1000)::bigint::text AS now')).rows
|
|
if(clock.length!==1)throw Error('Preflight terminal server clock unavailable')
|
|
liveValidity(current.policy,Number(clock[0].now))
|
|
if(approval.expiresAt<=Date.now())throw Error('Preflight approval expired before terminal')
|
|
const terminal=await s.query('ROLLBACK');if(terminal.command!=='ROLLBACK')throw Error('Preflight terminal unconfirmed');s.begun(false)
|
|
return Object.freeze({state:'matched',epoch:current.epoch,headSha256:current.headSha256,notAfter:current.policy.notAfter,
|
|
sourceMatched:true,rolesMatched:true,catalogueMatched:true,platformProtectionVerified:initialSource?.platformProtectionVerified===true&&terminalSource?.platformProtectionVerified===true,writerAuthorized:false,generationAuthenticated:false,readyAuthorized:false})
|
|
})
|
|
}
|
|
async function boundedPending(filename){
|
|
const h=await open(filename,'r');try{const stat=await h.stat();if(!stat.isFile()||stat.size<1||stat.size>2048)throw Error('Bounded operator pending cursor required');const raw=await h.readFile();if(raw.length!==stat.size)throw Error('Pending cursor changed');const p=JSON.parse(raw.toString('utf8'));if(!bytes(p).equals(raw)||Object.keys(p).sort().join(',')!=='epoch,headSha256,installationId,policySha256,previousHeadSha256,publisherPid'||!(/^[1-9][0-9]*$/).test(p.epoch)||!Number.isSafeInteger(p.publisherPid))throw Error('Exact pending cursor refused');return p}finally{await h.close()}
|
|
}
|
|
async function reconcilePendingPublication(pool,{journalDirectory,deadlineAt=Date.now()+120000}){
|
|
const directory=await realpath(journalDirectory),filename=path.join(directory,'pending-publication.json')
|
|
let proposal;try{proposal=await boundedPending(filename)}catch(error){if(error.code==='ENOENT')return {state:'idle'};throw error}
|
|
const result=await inspectPublication(pool,proposal,{deadlineAt})
|
|
const after=await boundedPending(filename);if(canonical(after)!==canonical(proposal))throw Error('Pending cursor changed during protected inspection')
|
|
await unlink(filename)
|
|
return {...result,proposal}
|
|
}
|
|
return Object.freeze({
|
|
preflightCurrentProfile,
|
|
publishCurrentProfile:(pool,options)=>withJournalLease(options.journalDirectory,()=>publishCurrentProfile(pool,options)),
|
|
inspectPublication,
|
|
reconcilePendingPublication:(pool,options)=>withJournalLease(options.journalDirectory,()=>reconcilePendingPublication(pool,options)),
|
|
})
|
|
}
|