Implemented the next deposit-observation step without activating live deposits
This commit is contained in:
Executable
+168
@@ -0,0 +1,168 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import type { Pool, PoolClient } from 'pg';
|
||||
import { z } from 'zod';
|
||||
import { XMR_RPC_UINT64_MAX } from '../../shared/xmr-amount.ts';
|
||||
import { CUSTODY_WRITE_LOCK } from './postgres-schema.ts';
|
||||
import { validateXmrAddress } from './xmr.ts';
|
||||
import type { MoneroIncomingObservation, MoneroObservationPage } from './monero-observer.ts';
|
||||
|
||||
const walletFingerprintSchema = z.string().regex(/^[a-f0-9]{64}$/);
|
||||
const scanInputSchema = z.object({
|
||||
scanId: z.string().uuid(),
|
||||
walletFingerprint: walletFingerprintSchema,
|
||||
expectedScanHeight: z.number().int().nonnegative().safe(),
|
||||
scannedAt: z.number().int().nonnegative().safe(),
|
||||
page: z.object({
|
||||
network: z.literal('mainnet'), walletHeight: z.number().int().nonnegative().safe(),
|
||||
fromHeight: z.number().int().nonnegative().safe(), throughHeight: z.number().int().nonnegative().safe(),
|
||||
primaryAddress: z.string().length(95), incoming: z.array(z.unknown()).max(4096),
|
||||
}).strict(),
|
||||
}).strict();
|
||||
const observationSchema = z.object({
|
||||
observationId: z.string().min(1).max(160), transactionId: z.string().regex(/^[a-f0-9]{64}$/),
|
||||
address: z.string().length(95),
|
||||
amountAtomicXmr: z.string().regex(/^[1-9]\d{0,19}$/),
|
||||
height: z.number().int().positive().safe(), confirmations: z.number().int().positive().safe(),
|
||||
timestamp: z.number().int().nonnegative().safe(), unlockTime: z.string().regex(/^(0|[1-9]\d{0,19})$/),
|
||||
accountIndex: z.number().int().nonnegative().safe(), addressIndex: z.number().int().nonnegative().safe(),
|
||||
locked: z.boolean(), doubleSpendSeen: z.boolean(),
|
||||
}).strict();
|
||||
const MAX_SCAN_RANGE = 100_000;
|
||||
const MAX_OVERLAP = 10_000;
|
||||
|
||||
export type PersistObservationScanInput = {
|
||||
scanId: string;
|
||||
walletFingerprint: string;
|
||||
expectedScanHeight: number;
|
||||
scannedAt: number;
|
||||
page: MoneroObservationPage;
|
||||
};
|
||||
export type PersistObservationScanResult = {
|
||||
scanId: string;
|
||||
scanHeight: number;
|
||||
observed: number;
|
||||
changed: number;
|
||||
reappeared: number;
|
||||
notSeen: number;
|
||||
replayed: boolean;
|
||||
};
|
||||
|
||||
function digest(value: unknown): string {
|
||||
return createHash('sha256').update(JSON.stringify(value)).digest('hex');
|
||||
}
|
||||
function rowDigest(observation: MoneroIncomingObservation): string {
|
||||
return digest([observation.observationId, observation.transactionId, observation.amountAtomicXmr,
|
||||
observation.height, observation.timestamp, observation.unlockTime,
|
||||
observation.accountIndex, observation.addressIndex, observation.locked, observation.doubleSpendSeen]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist one complete wallet-RPC scan page as observation-only state. This function does not
|
||||
* create customer credits or financial journals. A missing transfer produces `not-seen`, not
|
||||
* an automatic debit/reversal; operators must reconcile Monero reorgs before any later credit design.
|
||||
*/
|
||||
export async function persistMoneroObservationScan(pool: Pool, input: PersistObservationScanInput): Promise<PersistObservationScanResult> {
|
||||
const parsed = scanInputSchema.parse(input);
|
||||
const page = parsed.page;
|
||||
if (page.fromHeight > page.throughHeight || page.throughHeight !== page.walletHeight ||
|
||||
page.throughHeight - page.fromHeight > MAX_SCAN_RANGE || parsed.expectedScanHeight - page.fromHeight > MAX_OVERLAP ||
|
||||
page.fromHeight > parsed.expectedScanHeight || page.throughHeight < parsed.expectedScanHeight) {
|
||||
throw new Error('Monero observation scan range is not contiguous with its durable cursor');
|
||||
}
|
||||
await validateXmrAddress(page.primaryAddress);
|
||||
const observations = page.incoming.map(value => observationSchema.parse(value));
|
||||
const ids = new Set<string>();
|
||||
for (const observation of observations) {
|
||||
await validateXmrAddress(observation.address);
|
||||
if (observation.observationId !== `${observation.transactionId}:${observation.accountIndex}:${observation.addressIndex}` ||
|
||||
BigInt(observation.amountAtomicXmr) > XMR_RPC_UINT64_MAX || observation.height < page.fromHeight ||
|
||||
observation.height > page.throughHeight || observation.confirmations !== page.walletHeight - observation.height + 1 ||
|
||||
(observation.locked && observation.confirmations <= 0)) throw new Error('Monero observation is inconsistent with its scan page');
|
||||
if (ids.has(observation.observationId)) throw new Error('Duplicate Monero observation identity');
|
||||
ids.add(observation.observationId);
|
||||
}
|
||||
const pageDigest = digest([parsed.walletFingerprint, page.fromHeight, page.throughHeight,
|
||||
observations.map(row => [row.observationId, rowDigest(row)]).sort((a, b) => String(a[0]).localeCompare(String(b[0])))]);
|
||||
|
||||
const client = await pool.connect();
|
||||
try {
|
||||
await client.query('BEGIN');
|
||||
await client.query('SET LOCAL statement_timeout=\'10s\'; SET LOCAL lock_timeout=\'3s\'; SET LOCAL idle_in_transaction_session_timeout=\'10s\'' );
|
||||
await client.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
const priorScan = await client.query(`SELECT payload_digest,through_height FROM wk_custody.monero_observation_scans
|
||||
WHERE scan_id=$1`, [parsed.scanId]);
|
||||
if (priorScan.rowCount) {
|
||||
if (priorScan.rows[0].payload_digest !== pageDigest) throw new Error('Monero scan ID was reused with different observations');
|
||||
await client.query('COMMIT');
|
||||
return { scanId: parsed.scanId, scanHeight: Number(priorScan.rows[0].through_height), observed: observations.length,
|
||||
changed: 0, reappeared: 0, notSeen: 0, replayed: true };
|
||||
}
|
||||
let state = await client.query(`SELECT network,scan_height FROM wk_custody.monero_observer_state
|
||||
WHERE wallet_fingerprint=$1 FOR UPDATE`, [parsed.walletFingerprint]);
|
||||
if (!state.rowCount) {
|
||||
if (parsed.expectedScanHeight !== 0) throw new Error('First Monero observation scan must start from wallet history height zero');
|
||||
await client.query(`INSERT INTO wk_custody.monero_observer_state(wallet_fingerprint,network,scan_height,updated_at)
|
||||
VALUES($1,'mainnet',0,$2)`, [parsed.walletFingerprint, parsed.scannedAt]);
|
||||
state = await client.query(`SELECT network,scan_height FROM wk_custody.monero_observer_state
|
||||
WHERE wallet_fingerprint=$1 FOR UPDATE`, [parsed.walletFingerprint]);
|
||||
}
|
||||
if (state.rows[0].network !== 'mainnet' || Number(state.rows[0].scan_height) !== parsed.expectedScanHeight) {
|
||||
throw new Error('Monero observer cursor changed; reread durable state before retry');
|
||||
}
|
||||
|
||||
let changed = 0, reappeared = 0;
|
||||
for (const observation of observations) {
|
||||
const current = await client.query(`SELECT payload_digest,currently_present,first_observed_at
|
||||
FROM wk_custody.monero_transfer_observations
|
||||
WHERE wallet_fingerprint=$1 AND observation_id=$2 FOR UPDATE`, [parsed.walletFingerprint, observation.observationId]);
|
||||
const payloadDigest = rowDigest(observation);
|
||||
const eventType = !current.rowCount ? 'seen' : !current.rows[0].currently_present ? 'reappeared'
|
||||
: current.rows[0].payload_digest !== payloadDigest ? 'changed' : null;
|
||||
await client.query(`INSERT INTO wk_custody.monero_transfer_observations(
|
||||
wallet_fingerprint,observation_id,txid,monero_address,account_index,address_index,amount_atomic_xmr,block_height,
|
||||
wallet_height,confirmations,timestamp,unlock_time,locked,double_spend_seen,payload_digest,
|
||||
currently_present,last_seen_scan_id,first_observed_at,last_observed_at)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,true,$16,$17,$17)
|
||||
ON CONFLICT(wallet_fingerprint,observation_id) DO UPDATE SET
|
||||
monero_address=excluded.monero_address,
|
||||
amount_atomic_xmr=excluded.amount_atomic_xmr,block_height=excluded.block_height,wallet_height=excluded.wallet_height,
|
||||
confirmations=excluded.confirmations,timestamp=excluded.timestamp,unlock_time=excluded.unlock_time,
|
||||
locked=excluded.locked,double_spend_seen=excluded.double_spend_seen,payload_digest=excluded.payload_digest,
|
||||
currently_present=true,last_seen_scan_id=excluded.last_seen_scan_id,last_observed_at=excluded.last_observed_at`,
|
||||
[parsed.walletFingerprint, observation.observationId, observation.transactionId, observation.address,
|
||||
observation.accountIndex, observation.addressIndex, observation.amountAtomicXmr, observation.height, page.walletHeight,
|
||||
observation.confirmations, observation.timestamp, observation.unlockTime, observation.locked,
|
||||
observation.doubleSpendSeen, payloadDigest, parsed.scanId,
|
||||
current.rowCount ? Number(current.rows[0].first_observed_at) : parsed.scannedAt]);
|
||||
if (eventType) {
|
||||
await client.query(`INSERT INTO wk_custody.monero_observation_events
|
||||
(wallet_fingerprint,observation_id,scan_id,event_type,payload_digest,wallet_height,observed_at)
|
||||
VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT DO NOTHING`,
|
||||
[parsed.walletFingerprint, observation.observationId, parsed.scanId, eventType, payloadDigest, page.walletHeight, parsed.scannedAt]);
|
||||
if (eventType === 'changed') changed++;
|
||||
if (eventType === 'reappeared') reappeared++;
|
||||
}
|
||||
}
|
||||
|
||||
const missing = await client.query(`SELECT observation_id,payload_digest FROM wk_custody.monero_transfer_observations
|
||||
WHERE wallet_fingerprint=$1 AND currently_present=true AND block_height BETWEEN $2 AND $3
|
||||
AND last_seen_scan_id<>$4 FOR UPDATE`, [parsed.walletFingerprint, page.fromHeight, page.throughHeight, parsed.scanId]);
|
||||
for (const row of missing.rows) {
|
||||
await client.query(`UPDATE wk_custody.monero_transfer_observations SET currently_present=false,last_seen_scan_id=$3
|
||||
WHERE wallet_fingerprint=$1 AND observation_id=$2`, [parsed.walletFingerprint, row.observation_id, parsed.scanId]);
|
||||
await client.query(`INSERT INTO wk_custody.monero_observation_events
|
||||
(wallet_fingerprint,observation_id,scan_id,event_type,payload_digest,wallet_height,observed_at)
|
||||
VALUES($1,$2,$3,'not-seen',$4,$5,$6) ON CONFLICT DO NOTHING`,
|
||||
[parsed.walletFingerprint, row.observation_id, parsed.scanId, row.payload_digest, page.walletHeight, parsed.scannedAt]);
|
||||
}
|
||||
await client.query(`INSERT INTO wk_custody.monero_observation_scans
|
||||
(scan_id,wallet_fingerprint,from_height,through_height,payload_digest,completed_at) VALUES($1,$2,$3,$4,$5,$6)`,
|
||||
[parsed.scanId, parsed.walletFingerprint, page.fromHeight, page.throughHeight, pageDigest, parsed.scannedAt]);
|
||||
await client.query(`UPDATE wk_custody.monero_observer_state SET scan_height=GREATEST(scan_height,$2),updated_at=$3
|
||||
WHERE wallet_fingerprint=$1`, [parsed.walletFingerprint, page.throughHeight, parsed.scannedAt]);
|
||||
await client.query('COMMIT');
|
||||
return { scanId: parsed.scanId, scanHeight: Math.max(parsed.expectedScanHeight, page.throughHeight),
|
||||
observed: observations.length, changed, reappeared, notSeen: missing.rowCount ?? 0, replayed: false };
|
||||
} catch (error) { await client.query('ROLLBACK').catch(() => {}); throw error; }
|
||||
finally { client.release(); }
|
||||
}
|
||||
@@ -1,6 +1,5 @@
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { lstat, readFile } from 'node:fs/promises';
|
||||
import { isIP } from 'node:net';
|
||||
import { z } from 'zod';
|
||||
import { parseAtomicXmr, XMR_RPC_UINT64_MAX } from '../../shared/xmr-amount.ts';
|
||||
import { validateXmrAddress } from './xmr.ts';
|
||||
@@ -23,6 +22,7 @@ const heightResultSchema = z.object({ height: integerString, status: z.literal('
|
||||
const addressResultSchema = z.object({ address: z.string().length(95), addresses: z.array(z.unknown()).optional(), status: z.literal('OK') }).passthrough();
|
||||
const incomingTransferSchema = z.object({
|
||||
txid: z.string().regex(/^[a-f0-9]{64}$/),
|
||||
address: z.string().length(95),
|
||||
amount: integerString,
|
||||
height: integerString,
|
||||
timestamp: integerString,
|
||||
@@ -39,6 +39,7 @@ const transfersResultSchema = z.object({
|
||||
export type MoneroIncomingObservation = {
|
||||
observationId: string;
|
||||
transactionId: string;
|
||||
address: string;
|
||||
amountAtomicXmr: string;
|
||||
height: number;
|
||||
confirmations: number;
|
||||
@@ -96,7 +97,7 @@ function validateEndpoint(value: string): URL {
|
||||
if (url.username || url.password || url.search || url.hash || url.pathname !== '/json_rpc') {
|
||||
throw new Error('Monero wallet-RPC endpoint must be a bare /json_rpc URL');
|
||||
}
|
||||
const loopback = ['localhost', '127.0.0.1', '::1', '[::1]'].includes(url.hostname) || isIP(url.hostname) === 6 && url.hostname === '[::1]';
|
||||
const loopback = url.hostname === '127.0.0.1' || url.hostname === '[::1]';
|
||||
if (url.protocol !== 'https:' && !(url.protocol === 'http:' && loopback)) {
|
||||
throw new Error('Wallet RPC requires HTTPS, except loopback-only HTTP');
|
||||
}
|
||||
@@ -179,19 +180,22 @@ export function createMoneroWalletObserver(config: z.infer<typeof configSchema>,
|
||||
const transferRaw = transfersResultSchema.parse(await rpc('get_transfers', {
|
||||
in: true, filter_by_height: true, min_height: fromHeight, max_height: walletHeight,
|
||||
}));
|
||||
const incoming = transferRaw.in.map(item => {
|
||||
if (transferRaw.in.length >= MAX_TRANSFERS) throw new Error('Monero transfer page reached its safety limit; refusing to advance scan cursor');
|
||||
const incoming = await Promise.all(transferRaw.in.map(async item => {
|
||||
const amount = BigInt(item.amount);
|
||||
if (amount <= 0n || amount > XMR_RPC_UINT64_MAX) throw new Error('Invalid incoming Monero amount');
|
||||
const address = item.address;
|
||||
await validateXmrAddress(address);
|
||||
const height = integer(item.height, BigInt(Number.MAX_SAFE_INTEGER), 'transfer height');
|
||||
if (height < fromHeight || height > walletHeight) throw new Error('Wallet returned transfer outside requested height range');
|
||||
const major = integer(item.subaddr_index.major, 0xffffffffn, 'account index');
|
||||
const minor = integer(item.subaddr_index.minor, 0xffffffffn, 'address index');
|
||||
return { observationId: `${item.txid}:${major}:${minor}`, transactionId: item.txid,
|
||||
return { observationId: `${item.txid}:${major}:${minor}`, transactionId: item.txid, address,
|
||||
amountAtomicXmr: amount.toString(), height, confirmations: walletHeight - height + 1,
|
||||
timestamp: integer(item.timestamp, BigInt(Number.MAX_SAFE_INTEGER), 'timestamp'),
|
||||
unlockTime: item.unlock_time, accountIndex: major, addressIndex: minor,
|
||||
locked: item.locked, doubleSpendSeen: item.double_spend_seen };
|
||||
});
|
||||
}));
|
||||
const unique = new Set(incoming.map(row => row.observationId));
|
||||
if (unique.size !== incoming.length) throw new Error('Duplicate Monero wallet output observation');
|
||||
return { network: 'mainnet', walletHeight, fromHeight, throughHeight: walletHeight, primaryAddress, incoming };
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import { createHash } from 'node:crypto';
|
||||
import type { Pool, PoolClient } from 'pg';
|
||||
|
||||
export const CUSTODY_PG_VERSION = 1;
|
||||
export const CUSTODY_PG_VERSION = 2;
|
||||
export const CUSTODY_WRITE_LOCK = '7140824361';
|
||||
export const CUSTODY_SCHEMA = `
|
||||
export const CUSTODY_SCHEMA_V1 = `
|
||||
CREATE SCHEMA IF NOT EXISTS wk_custody;
|
||||
REVOKE ALL ON SCHEMA wk_custody FROM PUBLIC;
|
||||
SET LOCAL search_path = wk_custody, pg_catalog;
|
||||
@@ -64,7 +64,72 @@ REVOKE ALL ON ALL TABLES IN SCHEMA wk_custody FROM PUBLIC;
|
||||
REVOKE ALL ON ALL SEQUENCES IN SCHEMA wk_custody FROM PUBLIC;
|
||||
REVOKE ALL ON ALL FUNCTIONS IN SCHEMA wk_custody FROM PUBLIC;
|
||||
`;
|
||||
export const CUSTODY_SCHEMA_DIGEST = createHash('sha256').update(CUSTODY_SCHEMA).digest('hex');
|
||||
export const CUSTODY_SCHEMA_V1_DIGEST = createHash('sha256').update(CUSTODY_SCHEMA_V1).digest('hex');
|
||||
export const CUSTODY_MIGRATION_V2 = `
|
||||
CREATE TABLE wk_custody.monero_observer_state(
|
||||
wallet_fingerprint text PRIMARY KEY CHECK(wallet_fingerprint ~ '^[a-f0-9]{64}$'),
|
||||
network text NOT NULL CHECK(network='mainnet'),
|
||||
scan_height bigint NOT NULL CHECK(scan_height>=0),
|
||||
updated_at bigint NOT NULL CHECK(updated_at>=0)
|
||||
);
|
||||
CREATE TABLE wk_custody.monero_transfer_observations(
|
||||
wallet_fingerprint text NOT NULL REFERENCES wk_custody.monero_observer_state(wallet_fingerprint),
|
||||
observation_id text NOT NULL,
|
||||
txid text NOT NULL CHECK(txid ~ '^[a-f0-9]{64}$'),
|
||||
monero_address text NOT NULL CHECK(length(monero_address)=95),
|
||||
account_index bigint NOT NULL CHECK(account_index>=0),
|
||||
address_index bigint NOT NULL CHECK(address_index>=0),
|
||||
amount_atomic_xmr numeric(20,0) NOT NULL CHECK(amount_atomic_xmr>0 AND amount_atomic_xmr<=18446744073709551615),
|
||||
block_height bigint NOT NULL CHECK(block_height>0),
|
||||
wallet_height bigint NOT NULL CHECK(wallet_height>=block_height),
|
||||
confirmations bigint NOT NULL CHECK(confirmations=wallet_height-block_height+1),
|
||||
timestamp bigint NOT NULL CHECK(timestamp>=0),
|
||||
unlock_time text NOT NULL CHECK(unlock_time ~ '^(0|[1-9][0-9]{0,19})$'),
|
||||
locked boolean NOT NULL,
|
||||
double_spend_seen boolean NOT NULL,
|
||||
payload_digest text NOT NULL CHECK(payload_digest ~ '^[a-f0-9]{64}$'),
|
||||
currently_present boolean NOT NULL DEFAULT true,
|
||||
last_seen_scan_id uuid NOT NULL,
|
||||
first_observed_at bigint NOT NULL CHECK(first_observed_at>=0),
|
||||
last_observed_at bigint NOT NULL CHECK(last_observed_at>=first_observed_at),
|
||||
PRIMARY KEY(wallet_fingerprint,observation_id),
|
||||
UNIQUE(wallet_fingerprint,txid,account_index,address_index)
|
||||
);
|
||||
CREATE INDEX monero_observations_height ON wk_custody.monero_transfer_observations(wallet_fingerprint,block_height,observation_id);
|
||||
CREATE TABLE wk_custody.monero_observation_events(
|
||||
sequence bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
|
||||
wallet_fingerprint text NOT NULL,
|
||||
observation_id text NOT NULL,
|
||||
scan_id uuid NOT NULL,
|
||||
event_type text NOT NULL CHECK(event_type IN ('seen','changed','reappeared','not-seen')),
|
||||
payload_digest text NOT NULL CHECK(payload_digest ~ '^[a-f0-9]{64}$'),
|
||||
wallet_height bigint NOT NULL CHECK(wallet_height>=0),
|
||||
observed_at bigint NOT NULL CHECK(observed_at>=0),
|
||||
UNIQUE(wallet_fingerprint,observation_id,scan_id,event_type),
|
||||
FOREIGN KEY(wallet_fingerprint,observation_id)
|
||||
REFERENCES wk_custody.monero_transfer_observations(wallet_fingerprint,observation_id)
|
||||
);
|
||||
CREATE TABLE wk_custody.monero_observation_scans(
|
||||
scan_id uuid PRIMARY KEY,
|
||||
wallet_fingerprint text NOT NULL REFERENCES wk_custody.monero_observer_state(wallet_fingerprint),
|
||||
from_height bigint NOT NULL CHECK(from_height>=0),
|
||||
through_height bigint NOT NULL CHECK(through_height>=from_height),
|
||||
payload_digest text NOT NULL CHECK(payload_digest ~ '^[a-f0-9]{64}$'),
|
||||
completed_at bigint NOT NULL CHECK(completed_at>=0)
|
||||
);
|
||||
CREATE INDEX monero_scans_wallet_height ON wk_custody.monero_observation_scans(wallet_fingerprint,through_height DESC);
|
||||
CREATE FUNCTION wk_custody.monero_observation_events_immutable() RETURNS trigger LANGUAGE plpgsql SET search_path=wk_custody,pg_catalog AS $$
|
||||
BEGIN RAISE EXCEPTION 'Monero observation event history is append-only'; END $$;
|
||||
CREATE TRIGGER monero_observation_events_no_update BEFORE UPDATE OR DELETE ON wk_custody.monero_observation_events
|
||||
FOR EACH ROW EXECUTE FUNCTION wk_custody.monero_observation_events_immutable();
|
||||
CREATE TRIGGER monero_observation_scans_no_update BEFORE UPDATE OR DELETE ON wk_custody.monero_observation_scans
|
||||
FOR EACH ROW EXECUTE FUNCTION wk_custody.monero_observation_events_immutable();
|
||||
`;
|
||||
export const CUSTODY_SCHEMA_V2_DIGEST = createHash('sha256')
|
||||
.update(`${CUSTODY_SCHEMA_V1}\n-- custody migration v2 --\n${CUSTODY_MIGRATION_V2}`).digest('hex');
|
||||
/** Retained alias for callers that need the initial schema definition in disposable fixtures. */
|
||||
export const CUSTODY_SCHEMA = CUSTODY_SCHEMA_V1;
|
||||
export const CUSTODY_SCHEMA_DIGEST = CUSTODY_SCHEMA_V2_DIGEST;
|
||||
|
||||
/** Explicit operator migration; the HTTP runtime never creates or upgrades schema. */
|
||||
export async function migrateCustody(pool: Pool): Promise<void> {
|
||||
@@ -73,20 +138,34 @@ export async function migrateCustody(pool: Pool): Promise<void> {
|
||||
await c.query('BEGIN');
|
||||
await c.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
|
||||
const exists = await c.query("SELECT to_regclass('wk_custody.schema_versions') AS name");
|
||||
if (exists.rows[0].name) await verifyCustodySchema(c);
|
||||
else {
|
||||
await c.query(CUSTODY_SCHEMA);
|
||||
await c.query('INSERT INTO wk_custody.schema_versions VALUES($1,$2)', [CUSTODY_PG_VERSION, CUSTODY_SCHEMA_DIGEST]);
|
||||
let current = 0;
|
||||
if (!exists.rows[0].name) {
|
||||
await c.query(CUSTODY_SCHEMA_V1);
|
||||
await c.query('INSERT INTO wk_custody.schema_versions VALUES(1,$1)', [CUSTODY_SCHEMA_V1_DIGEST]);
|
||||
current = 1;
|
||||
} else current = await verifyCustodySchemaHistory(c);
|
||||
if (current === 1) {
|
||||
await c.query(CUSTODY_MIGRATION_V2);
|
||||
await c.query('INSERT INTO wk_custody.schema_versions VALUES(2,$1)', [CUSTODY_SCHEMA_V2_DIGEST]);
|
||||
}
|
||||
if (current > CUSTODY_PG_VERSION) throw new Error('Custody PostgreSQL schema is newer than this application');
|
||||
await verifyCustodySchema(c);
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
export async function verifyCustodySchema(c: PoolClient): Promise<void> {
|
||||
const version = await verifyCustodySchemaHistory(c);
|
||||
if (version !== CUSTODY_PG_VERSION) throw new Error('Custody PostgreSQL schema migration required');
|
||||
}
|
||||
async function verifyCustodySchemaHistory(c: PoolClient): Promise<number> {
|
||||
const rows = await c.query('SELECT version,digest FROM wk_custody.schema_versions ORDER BY version');
|
||||
if (rows.rowCount !== 1 || rows.rows[0].version !== CUSTODY_PG_VERSION || rows.rows[0].digest !== CUSTODY_SCHEMA_DIGEST) {
|
||||
throw new Error('Custody PostgreSQL schema version/digest mismatch; explicit migration required');
|
||||
const expected = [CUSTODY_SCHEMA_V1_DIGEST, CUSTODY_SCHEMA_V2_DIGEST];
|
||||
if (!rows.rowCount || rows.rowCount > expected.length || rows.rows.some((row, index) =>
|
||||
Number(row.version) !== index + 1 || row.digest !== expected[index])) {
|
||||
throw new Error('Custody PostgreSQL schema version/digest history mismatch; explicit migration required');
|
||||
}
|
||||
return Number(rows.rows.at(-1).version);
|
||||
}
|
||||
|
||||
/** Role must already exist (operator creates credentials out-of-band); never accept SQL identifiers from HTTP. */
|
||||
@@ -108,7 +187,35 @@ export async function grantCustodyRuntime(pool: Pool, role: string): Promise<voi
|
||||
GRANT INSERT(id,kind,created) ON wk_custody.journals TO ${name};
|
||||
GRANT INSERT ON wk_custody.entries,wk_custody.receipts,wk_custody.credits,wk_custody.custody_meta TO ${name};
|
||||
GRANT INSERT,UPDATE ON wk_custody.listings,wk_custody.orders,wk_custody.withdrawals,wk_custody.outbox TO ${name};`);
|
||||
await c.query(`REVOKE ALL PRIVILEGES ON wk_custody.monero_observer_state,wk_custody.monero_transfer_observations,
|
||||
wk_custody.monero_observation_events,wk_custody.monero_observation_scans FROM ${name};
|
||||
REVOKE ALL PRIVILEGES ON SEQUENCE wk_custody.monero_observation_events_sequence_seq FROM ${name};`);
|
||||
await c.query('COMMIT');
|
||||
} catch (e) { await c.query('ROLLBACK').catch(() => {}); throw e; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
|
||||
/** Separate least-privilege role for private wallet observations; it cannot read balances/journals or post credits. */
|
||||
export async function grantMoneroObserverRuntime(pool: Pool, role: string, custodyRuntimeRole = 'custody_runtime'): Promise<void> {
|
||||
if (!/^[a-z][a-z0-9_]{0,62}$/.test(role)) throw new Error('Invalid observer role identifier');
|
||||
if (role === custodyRuntimeRole) throw new Error('Monero observer requires a separate role from custody runtime');
|
||||
const name = `"${role}"`;
|
||||
const c = await pool.connect();
|
||||
try {
|
||||
await c.query('BEGIN'); await verifyCustodySchema(c);
|
||||
const r = await c.query('SELECT rolsuper,rolcreatedb,rolcreaterole,rolreplication,rolbypassrls FROM pg_roles WHERE rolname=$1', [role]);
|
||||
if (!r.rowCount || Object.values(r.rows[0]).some(Boolean)) throw new Error('Observer role must exist without administrative privileges');
|
||||
const memberships = await c.query(`SELECT 1 FROM pg_auth_members WHERE member=(SELECT oid FROM pg_roles WHERE rolname=$1) LIMIT 1`, [role]);
|
||||
if (memberships.rowCount) throw new Error('Observer role must not inherit privileges through role membership');
|
||||
const owner = await c.query("SELECT pg_get_userbyid(nspowner) AS owner FROM pg_namespace WHERE nspname='wk_custody'");
|
||||
if (owner.rows[0].owner === role) throw new Error('Observer role must not own the custody schema');
|
||||
await c.query(`GRANT USAGE ON SCHEMA wk_custody TO ${name};
|
||||
REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA wk_custody FROM ${name};
|
||||
REVOKE ALL PRIVILEGES ON ALL SEQUENCES IN SCHEMA wk_custody FROM ${name};
|
||||
GRANT SELECT,INSERT,UPDATE ON wk_custody.monero_observer_state,wk_custody.monero_transfer_observations TO ${name};
|
||||
GRANT SELECT,INSERT ON wk_custody.monero_observation_events,wk_custody.monero_observation_scans TO ${name};
|
||||
GRANT USAGE,SELECT ON SEQUENCE wk_custody.monero_observation_events_sequence_seq TO ${name};`);
|
||||
await c.query('COMMIT');
|
||||
} catch (error) { await c.query('ROLLBACK').catch(() => {}); throw error; }
|
||||
finally { c.release(); }
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user