From ee5dc353307ff118855b0860730d6b6334465c03 Mon Sep 17 00:00:00 2001 From: Eric Meehan Date: Wed, 21 Oct 2026 14:09:55 -0400 Subject: [PATCH] Implemented the next deposit-observation step without activating live deposits --- docs/canonical-indexes.md | 2 +- docs/custodial-market-design.md | 2 +- docs/production-plan.md | 2 +- server/market/monero-observation-store.ts | 168 ++++++++++++++++++++++ server/market/monero-observer.ts | 14 +- server/market/postgres-schema.ts | 125 ++++++++++++++-- tests/monero-observer.test.ts | 19 ++- tests/postgres/custody.test.ts | 87 ++++++++++- 8 files changed, 395 insertions(+), 24 deletions(-) create mode 100755 server/market/monero-observation-store.ts diff --git a/docs/canonical-indexes.md b/docs/canonical-indexes.md index 9cf3ac9..abd1051 100755 --- a/docs/canonical-indexes.md +++ b/docs/canonical-indexes.md @@ -16,7 +16,7 @@ Status: implemented local infrastructure, **not full M2 acceptance or production ## Tests and measurements -The final Docker test-target build passed TypeScript/Vite compilation. All **182 offline tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart, backup rehearsal and four offline Monero observer contract/security tests. The focused market suite passed **27 tests**, and the disposable PostgreSQL suite passed **31/31** in the previous run. +The final Docker test-target build passed TypeScript/Vite compilation. All **183 offline tests passed, zero failures/skips**, including native RandomX multi-process HTTP/peer/restart, backup rehearsal and five offline Monero observer contract/security tests. The focused market suite passed **27 tests**. The disposable PostgreSQL suite most recently passed **33/33**, including v1→v2 schema migration and observation persistence. - indexed ownership/sequence versus full replay for transfers, multi-card trades and forks; - append/disconnect/reconnect ordering, side-branch/mempool exclusion, bounded defensive-copy pages and durable restart; diff --git a/docs/custodial-market-design.md b/docs/custodial-market-design.md index 62d9639..07c8ad8 100755 --- a/docs/custodial-market-design.md +++ b/docs/custodial-market-design.md @@ -19,7 +19,7 @@ The operator has authorized proceeding with implementation instead of the previo **Financial storage update:** [ADR 003](postgres-financial-storage.md) selects PostgreSQL and records the integrated asynchronous custody backend, restricted runtime/schema roles, deferred accounting constraints, explicit SQLite freeze/import, and full dump/restore rehearsal. SQLite remains an opt-in-compatible local sandbox default, not an automatic fallback. Real custody and production operating acceptance remain blocked. -**Read-only Monero observer prototype:** [server/market/monero-observer.ts](../server/market/monero-observer.ts) defines a narrow mainnet wallet-RPC client for primary-address network check and bounded incoming-transfer scans. It requires an explicit protected config file and observer opt-in, uses authenticated requests with strict HTTPS/loopback-only HTTP, no redirects, bounded response/deadline, JSON-RPC request-ID validation and lossless JSON integer handling. The module exposes no subaddress creation, credit posting, transfer construction, relay, sweep or withdrawal operation and is not wired into application startup/routes. Fixture tests cover uint64 preservation, malformed/oversized responses, wrong RPC IDs and unsafe endpoint schemes. No wallet endpoint, credentials or funds were accessed. This is preparatory read-only plumbing, not live deposit crediting or an approved mainnet switch. +**Read-only Monero observation groundwork:** [server/market/monero-observer.ts](../server/market/monero-observer.ts) defines a narrow mainnet wallet-RPC client for mainnet primary-address verification and bounded incoming-transfer scans. It requires explicit opt-in and a protected external config file, uses authenticated requests with strict HTTPS/literal-loopback-only HTTP, no redirects, bounded response/deadline, JSON-RPC request-ID validation and lossless JSON integer handling. Each observed transfer includes its validated destination address. It exposes no subaddress creation, credit posting, transfer construction, relay, sweep or withdrawal operation and is not wired into server startup/routes. [server/market/monero-observation-store.ts](../server/market/monero-observation-store.ts) persists scan manifests, exact incoming-transfer observations, cursor progress and append-only seen/changed/not-seen/reappeared events in PostgreSQL schema v2, restricted to a dedicated observer DB role separate from the custody API runtime. Confirmation growth updates a projection without emitting an event every poll. A missing output is only a review signal; no automatic journal posting, debit, reversal or customer credit exists. Schema v1→v2 is an explicit additive migration, not an HTTP-runtime migration. The 33-test PostgreSQL suite covers v1 upgrade preservation, least-privilege separation, scan replay/cursor fencing and reorg/reappearance observations. No wallet endpoint, credentials or funds were accessed. This is observation/audit groundwork, not a deposit-address allocator, live deposit processor, or approved mainnet switch. The application only calls read methods, but operators must still use a restricted view-only wallet RPC instance/account; the code cannot prove credential scope. Validation: TypeScript/Vite build and 144 offline tests passed in the serial native-RandomX test run; the initial parallel run encountered a peer-gossip timeout. Browser smoke verified a disposable wallet's signed test credit, withdrawal hold and cancellation restoring the exact balance, plus a 390px viewport without horizontal overflow. No real funds were used. Tests and remaining acceptance gates are below; this is not launch readiness. diff --git a/docs/production-plan.md b/docs/production-plan.md index 143c868..437a509 100755 --- a/docs/production-plan.md +++ b/docs/production-plan.md @@ -20,7 +20,7 @@ Status: active engineering plan; **not launch or payment authorization**. Update **Current design deliverable:** [custodial market design](custodial-market-design.md). Read the workstreams below with this update: pooled customer liabilities replace per-order incoming invoices and external sale payouts in the proposed architecture. Earlier trustless-first research gates are historical, not a prerequisite to investigating custody. Exact ledger, withdrawal, fee, custody/legal and host-attested NFT consensus rules still require approval and evidence; the existing mock domain cannot authorize real balance debits. -**Read-only XMR groundwork:** a disabled-by-default Monero wallet-RPC observer client is implemented for mainnet address verification and bounded incoming-transfer scans, with protected config-file loading, transport limits and lossless uint64 parsing. It is deliberately not started by the server and cannot create addresses, post credits or spend funds. It has offline fixture coverage only. No actual RPC endpoint, wallet credential, address allocation or deposit was configured. Remaining deposit milestone: backed-up non-reusable subaddress allocation, durable output observation/inclusion/reorg history, exact idempotent ledger credit rules, controlled testnet/stagenet rehearsal, and independent accounting/security review before any real deposit can be accepted. +**Read-only XMR groundwork:** a disabled-by-default Monero wallet-RPC observer client is implemented for mainnet address validation and bounded incoming scans; an explicit PostgreSQL v1→v2 migration adds durable scan cursors, exact observation records and append-only reorg/reappearance markers under a separate observer database role. Each record includes a validated destination address. The observer/store remain disconnected from startup/routes and cannot allocate deposit addresses, post credits, reverse balances or spend. Missing outputs create review evidence only. Offline fixtures cover strict RPC safety, scan retry idempotency and cursor fencing; the PostgreSQL suite passed 33/33, including upgrade and reorg-observation persistence. No live RPC, wallet credential or funds were used. Remaining deposit milestone: backed-up non-reusable subaddress allocation, binding each credited output to a registered beneficiary/address, authoritative chain inclusion and reorg policy, exact idempotent journal crediting plus audited reversal/hold handling, controlled stagenet rehearsal, and independent accounting/security review before real deposits. ## 1A. MVP release boundary — proposed safe first production-hosted release diff --git a/server/market/monero-observation-store.ts b/server/market/monero-observation-store.ts new file mode 100755 index 0000000..6a76f1b --- /dev/null +++ b/server/market/monero-observation-store.ts @@ -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 { + 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(); + 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(); } +} diff --git a/server/market/monero-observer.ts b/server/market/monero-observer.ts index a7daa59..1132a4b 100755 --- a/server/market/monero-observer.ts +++ b/server/market/monero-observer.ts @@ -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, 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 }; diff --git a/server/market/postgres-schema.ts b/server/market/postgres-schema.ts index ab23c71..742c8ff 100755 --- a/server/market/postgres-schema.ts +++ b/server/market/postgres-schema.ts @@ -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 { @@ -73,20 +138,34 @@ export async function migrateCustody(pool: Pool): Promise { 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 { + const version = await verifyCustodySchemaHistory(c); + if (version !== CUSTODY_PG_VERSION) throw new Error('Custody PostgreSQL schema migration required'); +} +async function verifyCustodySchemaHistory(c: PoolClient): Promise { 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 {}); 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 { + 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(); } +} diff --git a/tests/monero-observer.test.ts b/tests/monero-observer.test.ts index b0c02de..1fe84c8 100755 --- a/tests/monero-observer.test.ts +++ b/tests/monero-observer.test.ts @@ -1,7 +1,7 @@ import test from 'node:test'; import assert from 'node:assert/strict'; import { ADDITIONAL_MARKET_FEE } from '../shared/market.ts'; -import { createMoneroWalletObserver, type MoneroFixtureTransport } from '../server/market/monero-observer.ts'; +import { createMoneroWalletObserver, loadMoneroObserverConfig, type MoneroFixtureTransport } from '../server/market/monero-observer.ts'; const config = { network: 'mainnet' as const, url: 'http://127.0.0.1:18083/json_rpc', username: 'fixture', password: 'not-a-live-secret' }; function response(id: string, result: Record, status = 200, type = 'application/json') { @@ -23,15 +23,16 @@ function fakeTransport(incoming: Record[] = []): { transport: M } test('read-only Monero observer preserves uint64 amount text and returns observations without crediting balances', async () => { - const { transport, calls } = fakeTransport([{ txid: 'a'.repeat(64), amount: '9007199254740993', height: 91, + const { transport, calls } = fakeTransport([{ txid: 'a'.repeat(64), address: ADDITIONAL_MARKET_FEE.address, + amount: '9007199254740993', height: 91, timestamp: 1_800_000_000, unlock_time: 0, locked: false, double_spend_seen: false, - subaddr_index: { major: 0, minor: 7 }, address: ADDITIONAL_MARKET_FEE.address }]); + subaddr_index: { major: 0, minor: 7 } }]); const observer = createMoneroWalletObserver(config, transport); const page = await observer.scanIncoming(90); assert.equal(page.network, 'mainnet'); assert.equal(page.walletHeight, 100); assert.equal(page.primaryAddress, ADDITIONAL_MARKET_FEE.address); assert.equal(page.incoming.length, 1); - assert.deepEqual(page.incoming[0], { observationId: `${'a'.repeat(64)}:0:7`, transactionId: 'a'.repeat(64), + assert.deepEqual(page.incoming[0], { observationId: `${'a'.repeat(64)}:0:7`, transactionId: 'a'.repeat(64), address: ADDITIONAL_MARKET_FEE.address, amountAtomicXmr: '9007199254740993', height: 91, confirmations: 10, timestamp: 1_800_000_000, unlockTime: '0', accountIndex: 0, addressIndex: 7, locked: false, doubleSpendSeen: false }); assert.deepEqual(calls.map(call => call.method).sort(), ['get_address', 'get_height', 'get_transfers']); @@ -43,6 +44,7 @@ test('read-only Monero observer preserves uint64 amount text and returns observa test('observer refuses non-loopback plaintext, URL credentials and malformed height ranges', async () => { assert.throws(() => createMoneroWalletObserver({ ...config, url: 'http://wallet.example/json_rpc' }), /HTTPS/); + assert.throws(() => createMoneroWalletObserver({ ...config, url: 'http://localhost/json_rpc' }), /HTTPS/); assert.throws(() => createMoneroWalletObserver({ ...config, url: 'https://user:pass@wallet.example/json_rpc' }), /bare/); const { transport } = fakeTransport(); const observer = createMoneroWalletObserver(config, transport); await assert.rejects(observer.scanIncoming(-1), /height/); @@ -54,12 +56,12 @@ test('observer rejects unsafe amounts, duplicate observations, malformed transfe const body = JSON.parse(init.body as string) as { id: string; method: string }; if (body.method === 'get_height') return response(body.id, { height: 100, status: 'OK' }); if (body.method === 'get_address') return response(body.id, { address: ADDITIONAL_MARKET_FEE.address, status: 'OK' }); - return new Response(`{"jsonrpc":"2.0","id":"${body.id}","result":{"status":"OK","in":[{"txid":"${'b'.repeat(64)}","amount":18446744073709551616,"height":91,"timestamp":1,"unlock_time":0,"locked":false,"double_spend_seen":false,"subaddr_index":{"major":0,"minor":1}}]}}`, + return new Response(`{"jsonrpc":"2.0","id":"${body.id}","result":{"status":"OK","in":[{"txid":"${'b'.repeat(64)}","address":"${ADDITIONAL_MARKET_FEE.address}","amount":18446744073709551616,"height":91,"timestamp":1,"unlock_time":0,"locked":false,"double_spend_seen":false,"subaddr_index":{"major":0,"minor":1}}]}}`, { headers: { 'content-type': 'application/json' } }); }; await assert.rejects(createMoneroWalletObserver(config, unsafeTransport).scanIncoming(90), /amount/); - const duplicate = { txid: 'c'.repeat(64), amount: '1', height: 91, timestamp: 1, unlock_time: 0, + const duplicate = { txid: 'c'.repeat(64), address: ADDITIONAL_MARKET_FEE.address, amount: '1', height: 91, timestamp: 1, unlock_time: 0, locked: true, double_spend_seen: false, subaddr_index: { major: 0, minor: 1 } }; const duplicateFixture = fakeTransport([duplicate, duplicate]); await assert.rejects(createMoneroWalletObserver(config, duplicateFixture.transport).scanIncoming(90), /Duplicate/); @@ -86,3 +88,8 @@ test('observer refuses redirects, wrong content type and oversized RPC response error instanceof Error && !error.message.includes(config.password)); } }); + +test('Monero RPC configuration is opt-in and inaccessible without a protected external config file', async () => { + await assert.rejects(loadMoneroObserverConfig({}), /disabled/); + await assert.rejects(loadMoneroObserverConfig({ WIKIDECK_MONERO_OBSERVER: '1' }), /config file is required/); +}); diff --git a/tests/postgres/custody.test.ts b/tests/postgres/custody.test.ts index 2288539..e28d913 100755 --- a/tests/postgres/custody.test.ts +++ b/tests/postgres/custody.test.ts @@ -14,7 +14,8 @@ import { PostgresCustody, auditPostgres } from '../../server/market/postgres-cus import { CustodyIntegrityError } from '../../server/market/custody-integrity.ts'; import { resolveCustodyIncident } from '../../server/market/custody-incident.ts'; import { CustodyMarket } from '../../server/market/custody.ts'; -import { migrateCustody, grantCustodyRuntime } from '../../server/market/postgres-schema.ts'; +import { migrateCustody, grantCustodyRuntime, grantMoneroObserverRuntime, CUSTODY_SCHEMA_V1, CUSTODY_SCHEMA_V1_DIGEST } from '../../server/market/postgres-schema.ts'; +import { persistMoneroObservationScan } from '../../server/market/monero-observation-store.ts'; import { importSqliteCustody } from '../../server/market/custody-import.ts'; import { postgresConfig, custodyStorage } from '../../server/market/postgres-config.ts'; import { openCustodyStorage } from '../../server/market/custody-storage.ts'; @@ -22,6 +23,7 @@ import { createBackup, restoreBackup } from '../../server/recovery.ts'; import { createCoordinatedCustodyBackup, verifyCoordinatedCustodyBackup } from '../../server/coordinated-backup.ts'; import { CUSTODY_POLICY, custodyMessage, type CustodyAction, type CustodyCommand, type CustodyActivity, type CustodyListing, type CustodyOrder, type CustodyWithdrawal } from '../../shared/custody.ts'; +import { ADDITIONAL_MARKET_FEE } from '../../shared/market.ts'; import { NETWORK } from '../../shared/network.ts'; import { CHAIN_ID, GENESIS_HASH, address, sha256 } from '../../server/testnet/consensus.ts'; import { saleContext, termsForCustodyOrder, saleCandidateTransactionId, saleId } from '../../server/testnet/sale-candidate.ts'; @@ -173,6 +175,89 @@ test('explicit PostgreSQL migrations are idempotent and runtime role cannot alte assert.equal((await f.market.audit()).assetsAtomicXmr, '1000000000000'); }); +test('custody schema v1 upgrades additively to observation-only v2 without dropping existing metadata', async t => { + const admin = new Pool(config); + await admin.query('DROP SCHEMA IF EXISTS wk_custody CASCADE'); + const client = await admin.connect(); + try { + await client.query('BEGIN'); await client.query(CUSTODY_SCHEMA_V1); + await client.query('INSERT INTO wk_custody.schema_versions VALUES(1,$1)', [CUSTODY_SCHEMA_V1_DIGEST]); + await client.query('INSERT INTO wk_custody.custody_meta VALUES($1,$2)', ['v1-preserved-marker', 'existing-test-data']); + await client.query('COMMIT'); + } catch (error) { await client.query('ROLLBACK'); throw error; } + finally { client.release(); } + try { + await migrateCustody(admin); await migrateCustody(admin); + const versions = await admin.query('SELECT version FROM wk_custody.schema_versions ORDER BY version'); + assert.deepEqual(versions.rows.map(row => Number(row.version)), [1, 2]); + assert.equal((await admin.query("SELECT value FROM wk_custody.custody_meta WHERE key='v1-preserved-marker'")).rows[0].value, + 'existing-test-data'); + for (const table of ['monero_observer_state', 'monero_transfer_observations', 'monero_observation_events', 'monero_observation_scans']) { + assert.ok((await admin.query('SELECT to_regclass($1) AS name', [`wk_custody.${table}`])).rows[0].name); + } + } finally { await admin.end(); } +}); + +test('Monero scan persistence is idempotent, reorg-visible and never posts customer credit', async t => { + const f = await fixture(t); + await f.admin.query('CREATE ROLE monero_observer LOGIN'); + await grantMoneroObserverRuntime(f.admin, 'monero_observer'); + await assert.rejects(grantMoneroObserverRuntime(f.admin, 'custody_runtime'), /separate role/); + const runtime = new Pool({ ...f.runtime, user: 'monero_observer' }); + const custodyRuntime = new Pool(f.runtime), wallet = 'd'.repeat(64), scan1 = randomUUID(); + const txid = 'e'.repeat(64), observationId = `${txid}:0:4`; + const observation = { observationId, transactionId: txid, amountAtomicXmr: '9007199254740993', height: 95, + address: ADDITIONAL_MARKET_FEE.address, + confirmations: 6, timestamp: 1_800_000_000, unlockTime: '0', accountIndex: 0, addressIndex: 4, + locked: false, doubleSpendSeen: false }; + const page = { network: 'mainnet' as const, walletHeight: 100, fromHeight: 0, throughHeight: 100, + primaryAddress: ADDITIONAL_MARKET_FEE.address, incoming: [observation] }; + try { + await assert.rejects(custodyRuntime.query('SELECT * FROM wk_custody.monero_transfer_observations')); + await assert.rejects(runtime.query('SELECT * FROM wk_custody.accounts')); + const first = await persistMoneroObservationScan(runtime, { scanId: scan1, walletFingerprint: wallet, + expectedScanHeight: 0, scannedAt: 1_800_000_100, page }); + assert.equal(first.observed, 1); assert.equal(first.scanHeight, 100); assert.equal(first.replayed, false); + const replay = await persistMoneroObservationScan(runtime, { scanId: scan1, walletFingerprint: wallet, + expectedScanHeight: 0, scannedAt: 1_800_000_100, page }); + assert.equal(replay.replayed, true); + await assert.rejects(persistMoneroObservationScan(runtime, { scanId: scan1, walletFingerprint: wallet, + expectedScanHeight: 0, scannedAt: 1_800_000_100, page: { ...page, incoming: [] } }), /reused/); + + const confirmationRefresh = await persistMoneroObservationScan(runtime, { scanId: randomUUID(), walletFingerprint: wallet, + expectedScanHeight: 100, scannedAt: 1_800_000_150, + page: { ...page, fromHeight: 90, walletHeight: 105, throughHeight: 105, + incoming: [{ ...observation, confirmations: 11 }] } }); + assert.equal(confirmationRefresh.changed, 0, 'normal confirmation growth is a projection update, not an immutable event per poll'); + + const missing = await persistMoneroObservationScan(runtime, { scanId: randomUUID(), walletFingerprint: wallet, + expectedScanHeight: 105, scannedAt: 1_800_000_200, + page: { ...page, fromHeight: 90, walletHeight: 110, throughHeight: 110, incoming: [] } }); + assert.equal(missing.notSeen, 1); // Review signal only; never an automatic financial reversal. + const reappeared = await persistMoneroObservationScan(runtime, { scanId: randomUUID(), walletFingerprint: wallet, + expectedScanHeight: 110, scannedAt: 1_800_000_300, + page: { ...page, fromHeight: 90, walletHeight: 120, throughHeight: 120, + incoming: [{ ...observation, confirmations: 26 }] } }); + assert.equal(reappeared.reappeared, 1); + + const observed = await runtime.query(`SELECT amount_atomic_xmr::text AS amount,monero_address,currently_present FROM wk_custody.monero_transfer_observations + WHERE wallet_fingerprint=$1 AND observation_id=$2`, [wallet, observationId]); + assert.equal(observed.rows[0].amount, '9007199254740993'); assert.equal(observed.rows[0].currently_present, true); + assert.equal(observed.rows[0].monero_address, ADDITIONAL_MARKET_FEE.address); + const eventTypes = await runtime.query('SELECT event_type FROM wk_custody.monero_observation_events ORDER BY sequence'); + assert.deepEqual(eventTypes.rows.map(row => row.event_type), ['seen', 'not-seen', 'reappeared']); + await assert.rejects(runtime.query('UPDATE wk_custody.monero_observation_events SET event_type=\'seen\'')); + await assert.rejects(runtime.query('DELETE FROM wk_custody.monero_observation_scans')); + await assert.rejects(runtime.query('SELECT count(*) FROM wk_custody.journals'), /permission denied/); + assert.equal((await f.admin.query('SELECT count(*)::int AS n FROM wk_custody.journals')).rows[0].n, 0, + 'observation persistence must never create financial postings'); + await assert.rejects(persistMoneroObservationScan(runtime, { scanId: randomUUID(), walletFingerprint: wallet, + expectedScanHeight: 110, scannedAt: 1_800_000_400, + page: { ...page, fromHeight: 90, walletHeight: 121, throughHeight: 121, + incoming: [{ ...observation, confirmations: 27 }] } }), /cursor changed/); + } finally { await custodyRuntime.end(); await runtime.end(); } +}); + test('database rejects unbalanced, fractional, negative, historical append and oversized postings at commit', async t => { const f = await fixture(t); await f.execute(buyer, { type: 'mock-credit' }); const c = await f.admin.connect();