diff --git a/.changeset/persistent-cli-sessions.md b/.changeset/persistent-cli-sessions.md new file mode 100644 index 00000000..92912a60 --- /dev/null +++ b/.changeset/persistent-cli-sessions.md @@ -0,0 +1,5 @@ +--- +'mppx': patch +--- + +Added persistent CLI payment sessions with automatic reuse, inspection, and explicit closing. diff --git a/README.md b/README.md index 31d5e8b6..11262b71 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,8 @@ npx gitpick wevm/mppx/examples/charge ## CLI -`mppx` includes a basic CLI for making HTTP requests with automatic payment handling. +`mppx` includes a basic CLI for making HTTP requests with automatic payment handling. Tempo +session channels are retained and reused automatically until you close them. ```bash # create account - stored in keychain, autofunded on testnet @@ -105,8 +106,20 @@ mppx account create # make request - automatic payment handling, curl-like api mppx example.com + +# open another session instead of reusing the preferred channel +mppx example.com --session new + +# inspect and close retained sessions +mppx sessions list +mppx sessions view +mppx sessions close +mppx sessions close --all --yes ``` +`--session auto` is the default. Pass `new` to open another channel or a channel ID to select one +explicitly. + You can also install globally to use the `mppx` CLI from anywhere: ```bash diff --git a/src/cli/account.ts b/src/cli/account.ts index 5cfc3cdb..eb1d6763 100644 --- a/src/cli/account.ts +++ b/src/cli/account.ts @@ -5,6 +5,8 @@ import * as path from 'node:path' import { Errors } from 'incur' +import { isTempoAccount } from './utils.js' + const SERVICE_NAME = 'mppx' const defaultCommandTimeoutMs = 10_000 @@ -203,6 +205,48 @@ export function createKeychain(account = 'main') { } } +/** Resolves a local CLI signer together with its durable account reference. */ +export async function resolveLocalAccount(name?: string) { + const { privateKeyToAccount } = await import('viem/accounts') + + const envKey = process.env.MPPX_PRIVATE_KEY?.trim() + if (envKey) + return { + account: privateKeyToAccount(envKey as `0x${string}`), + source: 'environment' as const, + } + + const accountName = resolveAccountName(name) + const key = await createKeychain(accountName).get() + if (key) + return { + account: privateKeyToAccount(key as `0x${string}`), + accountName, + source: 'keychain' as const, + } + + throw new Error(`Account "${accountName}" not found.`) +} + +/** Resolves an account supported by persistent CLI sessions. */ +export async function resolvePersistentAccount(name?: string) { + const accountName = resolveAccountName(name) + if (!process.env.MPPX_PRIVATE_KEY?.trim() && isTempoAccount(accountName)) + throw new Errors.IncurError({ + code: 'UNSUPPORTED_ACCOUNT', + message: 'Persistent sessions require an mppx account or MPPX_PRIVATE_KEY.', + exitCode: 2, + }) + return resolveLocalAccount(name).catch((cause: unknown) => { + throw new Errors.IncurError({ + code: 'ACCOUNT_NOT_FOUND', + message: cause instanceof Error ? cause.message : 'No account found.', + exitCode: 69, + ...(cause instanceof Error && { cause }), + }) + }) +} + /** * Resolve a CLI account to a viem `LocalAccount`. * @@ -221,14 +265,5 @@ export function createKeychain(account = 'main') { * ``` */ export async function resolveAccount(name?: string) { - const { privateKeyToAccount } = await import('viem/accounts') - - const envKey = process.env.MPPX_PRIVATE_KEY?.trim() - if (envKey) return privateKeyToAccount(envKey as `0x${string}`) - - const accountName = resolveAccountName(name) - const key = await createKeychain(accountName).get() - if (key) return privateKeyToAccount(key as `0x${string}`) - - throw new Error(`Account "${accountName}" not found.`) + return (await resolveLocalAccount(name)).account } diff --git a/src/cli/cli.test.ts b/src/cli/cli.test.ts index fcc5b393..cfb3681f 100644 --- a/src/cli/cli.test.ts +++ b/src/cli/cli.test.ts @@ -32,6 +32,7 @@ import cli from './cli.js' const testPrivateKey = generatePrivateKey() const testAccount = privateKeyToAccount(testPrivateKey) const cliTestXdgDataHome = fs.mkdtempSync(path.join(os.tmpdir(), 'mppx-cli-xdg-')) +let cliServeInvocation = 0 const unfundedToken = '0x20c0000000000000000000000000000000000002' as Address const cliSessionFeePayerPolicy = { maxGas: 2_250_000n, @@ -47,7 +48,13 @@ async function serve(argv: string[], options?: { env?: Record = {} - const env = { XDG_DATA_HOME: cliTestXdgDataHome, ...options?.env } + const invocationHome = path.join(cliTestXdgDataHome, `serve-${cliServeInvocation++}`) + const env = { + XDG_CONFIG_HOME: path.join(invocationHome, 'config'), + XDG_DATA_HOME: cliTestXdgDataHome, + XDG_STATE_HOME: path.join(invocationHome, 'state'), + ...options?.env, + } for (const [key, value] of Object.entries(env)) { saved[key] = process.env[key] if (value === undefined) delete process.env[key] @@ -1049,10 +1056,10 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { try { const { output } = await serve( - [httpServer.url, '--rpc-url', rpcUrl, '-s', '-M', 'deposit=10'], + [httpServer.url, '--rpc-url', rpcUrl, '--include', '-M', 'deposit=10'], { env: { MPPX_PRIVATE_KEY: testPrivateKey } }, ) - expect(output).toContain('scraped-content') + expect(output).toBe('scraped-content') } finally { httpServer.close() } @@ -1133,71 +1140,108 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { } }) - test('bug: non-SSE open should not double-charge tick amount', { timeout: 120_000 }, async () => { - await fundAccount({ address: testAccount.address, token: Addresses.pathUsd }) - await fundAccount({ address: testAccount.address, token: asset }) - - const escrow = tip20ChannelEscrow - const store = Store.memory() - const tickAmount = '0.001' - const server = Mppx_server.create({ - methods: [ - tempo.session({ - account: accounts[0], - store, - getClient: () => client, - currency: asset, - escrowContract: escrow, - chainId: client.chain.id, - feePayer: true, - feePayerPolicy: cliSessionFeePayerPolicy, - }), - ], - realm: 'cli-test-double-charge', - secretKey: 'cli-test-secret-cli-test-secret-32', - }) + test( + 'selects retained non-SSE channels with auto, explicit, and new modes', + { timeout: 120_000 }, + async () => { + await fundAccount({ address: testAccount.address, token: Addresses.pathUsd }) + await fundAccount({ address: testAccount.address, token: asset }) - // Track voucher cumulative amounts from credential payloads - const voucherAmounts: string[] = [] + const escrow = tip20ChannelEscrow + const store = Store.memory() + const tickAmount = '0.001' + const server = Mppx_server.create({ + methods: [ + tempo.session({ + account: accounts[0], + store, + getClient: () => client, + currency: asset, + escrowContract: escrow, + chainId: client.chain.id, + feePayer: true, + feePayerPolicy: cliSessionFeePayerPolicy, + }), + ], + realm: 'cli-test-double-charge', + secretKey: 'cli-test-secret-cli-test-secret-32', + }) - const httpServer = await Http.createServer(async (req, res) => { - const authHeader = req.headers.authorization - if (authHeader) { - try { - const cred = Credential.deserialize(authHeader) - if (cred.payload.action === 'voucher' && 'cumulativeAmount' in cred.payload) { - voucherAmounts.push(cred.payload.cumulativeAmount) - } - } catch {} - } + const voucherAmounts: string[] = [] + const credentials: { action: string; channelId: string }[] = [] - const result = await toNodeListener( - server.session({ - amount: tickAmount, - recipient: accounts[0].address, - unitType: 'page', - }), - )(req, res) - if (result.status === 402) return - // Non-SSE: plain text response (not text/event-stream) - res.end('scraped-content') - }) + const httpServer = await Http.createServer(async (req, res) => { + const authHeader = req.headers.authorization + if (authHeader) { + try { + const cred = Credential.deserialize(authHeader) + credentials.push({ + action: cred.payload.action, + channelId: cred.payload.channelId, + }) + if (cred.payload.action === 'voucher' && 'cumulativeAmount' in cred.payload) { + voucherAmounts.push(cred.payload.cumulativeAmount) + } + } catch {} + } - try { - await serve([httpServer.url, '--rpc-url', rpcUrl, '-s', '-M', 'deposit=10'], { - env: { MPPX_PRIVATE_KEY: testPrivateKey }, + const result = await toNodeListener( + server.session({ + amount: tickAmount, + recipient: accounts[0].address, + unitType: 'page', + }), + )(req, res) + if (result.status === 402) return + // Non-SSE: plain text response (not text/event-stream) + res.end('scraped-content') }) - // No follow-up voucher should be sent after a non-SSE open. - // The open credential already paid for this unit, so the CLI - // should NOT send a redundant voucher that would double-charge. - expect(voucherAmounts.length).toBe(0) - } finally { - httpServer.close() - } - }) + try { + const stateHome = fs.mkdtempSync(path.join(cliTestXdgDataHome, 'reuse-state-')) + const env = { MPPX_PRIVATE_KEY: testPrivateKey, XDG_STATE_HOME: stateHome } + const runRequest = async (argv: string[]) => { + const result = await serve(argv, { env }) + expect(result.exitCode, `${result.output}\n${result.stderr}`).toBeUndefined() + } + await runRequest([httpServer.url, '--rpc-url', rpcUrl, '-s', '-M', 'deposit=10']) + const retainedChannelId = credentials.find(({ action }) => action === 'open')!.channelId + await runRequest([httpServer.url, '--rpc-url', rpcUrl, '-s', '-M', 'deposit=10']) + await runRequest([ + httpServer.url, + '--rpc-url', + rpcUrl, + '--session', + retainedChannelId, + '-s', + '-M', + 'deposit=10', + ]) + await runRequest([ + httpServer.url, + '--rpc-url', + rpcUrl, + '--session', + 'new', + '-s', + '-M', + 'deposit=10', + ]) + + const payments = credentials.filter(({ action }) => ['open', 'voucher'].includes(action)) + expect(voucherAmounts).toEqual(['2000', '3000']) + expect(payments.map(({ action }) => action)).toEqual(['open', 'voucher', 'voucher', 'open']) + expect(new Set(payments.slice(0, 3).map(({ channelId }) => channelId))).toEqual( + new Set([retainedChannelId]), + ) + expect(payments[3]?.channelId).not.toBe(retainedChannelId) + } finally { + httpServer.close() + } + }, + ) - test('bug: closeChannel sends action "close" not "voucher"', { timeout: 120_000 }, async () => { + test('retains a session until `sessions close` is invoked', { timeout: 120_000 }, async () => { await fundAccount({ address: testAccount.address, token: Addresses.pathUsd }) await fundAccount({ address: testAccount.address, token: asset }) @@ -1212,7 +1256,7 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { currency: asset, escrowContract: escrow, chainId: client.chain.id, - feePayer: true, + feePayer: accounts[1], feePayerPolicy: cliSessionFeePayerPolicy, }), ], @@ -1220,8 +1264,8 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { secretKey: 'cli-test-secret-cli-test-secret-32', }) - // Track the credential payload action from the close request const credentialActions: string[] = [] + let channelId: string | undefined const httpServer = await Http.createServer(async (req, res) => { // Capture credential action from every request with Authorization header @@ -1230,6 +1274,7 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { try { const cred = Credential.deserialize(authHeader) credentialActions.push(cred.payload.action) + channelId = cred.payload.channelId } catch {} } @@ -1245,13 +1290,31 @@ describe('session multi-fetch (examples/session/multi-fetch)', () => { }) try { + const stateHome = fs.mkdtempSync(path.join(cliTestXdgDataHome, 'close-state-')) + const env = { MPPX_PRIVATE_KEY: testPrivateKey, XDG_STATE_HOME: stateHome } await serve([httpServer.url, '--rpc-url', rpcUrl, '-s', '-M', 'deposit=10'], { - env: { MPPX_PRIVATE_KEY: testPrivateKey }, + env, + }) + + expect(credentialActions).toEqual(['open']) + expect(channelId).toMatch(/^0x[0-9a-f]{64}$/) + + const listed = await serve(['sessions', 'list', '--json'], { env }) + expect(JSON.parse(listed.output).sessions).toEqual([ + expect.objectContaining({ channelId, status: 'open' }), + ]) + + const closed = await serve(['sessions', 'close', channelId!, '--rpc-url', rpcUrl, '--json'], { + env, }) + expect(closed.exitCode, `${closed.output}\n${closed.stderr}`).toBeUndefined() + expect(JSON.parse(closed.output)).toEqual( + expect.objectContaining({ channelId, status: 'closed' }), + ) + expect(credentialActions.at(-1)).toBe('close') - // The last credential sent should be the close request with action: 'close' - const lastAction = credentialActions[credentialActions.length - 1] - expect(lastAction).toBe('close') + const empty = await serve(['sessions', 'list', '--json'], { env }) + expect(JSON.parse(empty.output)).toEqual({ sessions: [] }) } finally { httpServer.close() } @@ -1298,7 +1361,9 @@ describe('session sse (examples/session/sse)', () => { secretKey: 'cli-test-secret-cli-test-secret-32', }) + const paidAcceptHeaders: (string | undefined)[] = [] const httpServer = await Http.createServer(async (req, res) => { + if (req.headers.authorization) paidAcceptHeaders.push(req.headers.accept) const result = await toNodeListener( server.session({ amount: '0.001', @@ -1322,10 +1387,12 @@ describe('session sse (examples/session/sse)', () => { }) try { - const { output } = await serve([httpServer.url, '--rpc-url', rpcUrl, '-M', 'deposit=10'], { - env: { MPPX_PRIVATE_KEY: testPrivateKey }, - }) + const { output } = await serve( + [httpServer.url, '--rpc-url', rpcUrl, '-H', 'Accept: application/json', '-M', 'deposit=10'], + { env: { MPPX_PRIVATE_KEY: testPrivateKey } }, + ) expect(output.trim()).toBe('Hello world!') + expect(paidAcceptHeaders).toEqual(['application/json']) } finally { httpServer.close() } @@ -1836,8 +1903,36 @@ test('mppx --help', async () => { expect(output).toContain('mppx') expect(output).toContain('') expect(output).toContain('account') + expect(output).toContain('sessions') expect(output).toContain('services') expect(output).toContain('sign') + expect(output).toContain('--session') +}) + +test('mppx sessions --help', async () => { + const { output } = await serve(['sessions', '--help']) + expect(output).toContain('close Cooperatively close') + expect(output).toContain('list List persistent') + expect(output).toContain('view View a persistent') +}) + +test('mppx sessions close --help', async () => { + const { output } = await serve(['sessions', 'close', '--help']) + expect(output).toContain('--all') + expect(output).toContain('--yes') + expect(output).toContain('--url') +}) + +test('mppx sessions close --all requires confirmation', async () => { + const { exitCode, output } = await serve(['sessions', 'close', '--all']) + expect(exitCode).toBe(2) + expect(output).toContain('requires --yes') +}) + +test('mppx sessions close --all returns an empty structured result', async () => { + const { exitCode, output } = await serve(['sessions', 'close', '--all', '--yes', '--json']) + expect(exitCode).toBeUndefined() + expect(JSON.parse(output)).toEqual({ closed: [], failed: [] }) }) describe('account fund help', () => { diff --git a/src/cli/cli.ts b/src/cli/cli.ts index 48176e1a..ba081268 100755 --- a/src/cli/cli.ts +++ b/src/cli/cli.ts @@ -13,6 +13,7 @@ import { normalizeHeaders } from '../client/internal/Fetch.js' import * as Mppx from '../client/Mppx.js' import * as Constants from '../Constants.js' import { validate as validateDiscovery } from '../discovery/Validate.js' +import { isTempoSessionChallenge } from '../tempo/session/client/Transports.js' import { createDefaultStore, createKeychain, resolveAccountName } from './account.js' import { loadConfig, resolveAcceptPayment, selectChallenge } from './internal.js' import type { Plugin } from './plugins/plugin.js' @@ -21,6 +22,9 @@ import { readTempoKeystore, resolveTempoAccount, } from './plugins/tempo.js' +import sessions, { sessionCommandError } from './sessions/commands.js' +import { runPersistentSessionRequest } from './sessions/request.js' +import { createSessionRegistry } from './sessions/store.js' import { chainName, confirm, @@ -244,6 +248,15 @@ const cli = Cli.create('mppx', { .string() .optional() .describe('RPC endpoint, defaults to public RPC for chain (env: MPPX_RPC_URL)'), + session: z + .string() + .optional() + .default('auto') + .refine( + (value) => value === 'auto' || value === 'new' || /^0x[0-9a-fA-F]{64}$/.test(value), + 'Expected auto, new, or a 32-byte channel ID', + ) + .describe('Session selection: auto, new, or channel ID'), silent: z.boolean().default(false).describe('Silent mode (suppress progress and info)'), slippage: z.number().optional().describe('Tempo auto-swap max slippage percentage'), userAgent: z @@ -347,6 +360,9 @@ const cli = Cli.create('mppx', { } try { + const methodOpts = parseMethodOpts(c.options.methodOpt) + const sessionRegistry = createSessionRegistry() + const init: RequestInit = { redirect: c.options.location ? 'follow' : 'manual' } if (c.options.jsonBody) { init.body = c.options.jsonBody @@ -368,6 +384,12 @@ const cli = Cli.create('mppx', { if (c.options.verbose >= 2) printRequestHeaders(url, init, info) const challengeResponse = await targetFetch(fetchUrl, init) if (challengeResponse.status !== 402) { + if (c.options.session !== 'auto') + return c.error({ + code: 'UNSUPPORTED_SESSION', + message: '--session requires a tempo/session payment challenge.', + exitCode: 2, + }) if (c.options.fail && challengeResponse.status >= 400) return c.error({ code: 'HTTP_ERROR', @@ -379,7 +401,6 @@ const cli = Cli.create('mppx', { return } - const methodOpts = parseMethodOpts(c.options.methodOpt) const offeredChallenges = Challenge.fromResponseList(challengeResponse) const currencyChallenges = filterChallengesByCurrency(offeredChallenges, c.options.currency) if (c.options.currency && currencyChallenges.length === 0) { @@ -446,7 +467,12 @@ const cli = Cli.create('mppx', { // Display challenge const shownKeys = new Set() { - printResponseHeaders(challengeResponse, headerOpts) + printResponseHeaders( + challengeResponse, + challenge.method === 'tempo' && challenge.intent === 'session' && c.options.include + ? { ...headerOpts, include: false, verbose: Math.max(headerOpts.verbose, 2) } + : headerOpts, + ) const challengeRows = (() => { const skip = new Set(['id', 'request']) @@ -517,6 +543,44 @@ const cli = Cli.create('mppx', { } } + const persistentSessionAccount = + process.env.MPPX_PRIVATE_KEY?.trim() || + !isTempoAccount(resolveAccountName(c.options.account)) + if (isTempoSessionChallenge(challenge) && persistentSessionAccount) { + try { + await runPersistentSessionRequest({ + challenge, + challengeResponse: selectedChallengeResponse, + endpoint: url, + fetch: targetFetch, + fetchInput: fetchUrl, + init, + info, + methodOptions: methodOpts, + options: { + account: c.options.account, + fail: c.options.fail, + include: c.options.include, + network: c.options.network, + rpcUrl: c.options.rpcUrl, + session: c.options.session, + silent: c.options.silent, + verbose: c.options.verbose, + }, + registry: sessionRegistry, + }) + } catch (error) { + return sessionCommandError(error, 'SESSION_REQUEST_FAILED') + } + return + } + if (c.options.session !== 'auto') + return c.error({ + code: 'UNSUPPORTED_SESSION', + message: '--session requires a tempo/session payment challenge.', + exitCode: 2, + }) + // Create credential let credential: string if (pluginResult?.createCredential) @@ -1578,6 +1642,7 @@ cli.command(account) cli.command(discover) cli.command(init) cli.command(services) +cli.command(sessions) cli.command(sign) cli.command(validate) diff --git a/src/cli/mcp.test.ts b/src/cli/mcp.test.ts index 4bef209d..1636a4fc 100644 --- a/src/cli/mcp.test.ts +++ b/src/cli/mcp.test.ts @@ -193,6 +193,9 @@ test('tools/list exposes mppx commands with input and output schemas', async () 'services_endpoints', 'services_list', 'services_show', + 'sessions_close', + 'sessions_list', + 'sessions_view', 'sign', 'validate', ]) @@ -208,6 +211,14 @@ test('tools/list exposes mppx commands with input and output schemas', async () type: 'object', }), ) + expect( + tools.find((tool: { name: string }) => tool.name === 'sessions_list').outputSchema, + ).toEqual( + expect.objectContaining({ + properties: expect.objectContaining({ sessions: expect.any(Object) }), + type: 'object', + }), + ) expect(client.nonJsonLines).toEqual([]) }) diff --git a/src/cli/sessions/Manager.test.ts b/src/cli/sessions/Manager.test.ts new file mode 100644 index 00000000..76e5ec5e --- /dev/null +++ b/src/cli/sessions/Manager.test.ts @@ -0,0 +1,249 @@ +import { createClient, custom, type Hex } from 'viem' +import { privateKeyToAccount } from 'viem/accounts' +import { describe, expect, test, vi } from 'vp/test' + +import * as Challenge from '../../Challenge.js' +import * as Constants from '../../Constants.js' +import * as Credential from '../../Credential.js' +import type { ChannelEntry } from '../../tempo/session/client/ChannelOps.js' +import { entryKey, type ChannelStore } from '../../tempo/session/client/ChannelStore.js' +import type { TempoSessionChallenge } from '../../tempo/session/client/Transports.js' +import * as Channel from '../../tempo/session/precompile/Channel.js' +import { + createSessionReceipt, + serializeSessionReceipt, + tip20ChannelEscrow, + type ChannelDescriptor, + type SessionCredentialPayload, +} from '../../tempo/session/precompile/Protocol.js' +import type { SessionSnapshot } from '../../tempo/session/Snapshot.js' +import { closeWithSessionManager } from './Manager.js' + +const account = privateKeyToAccount( + '0xac0974bec39a17e36ba6a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80', +) +const client = createClient({ + account, + chain: { id: 4217 } as never, + transport: custom({ + async request(args) { + throw new Error(`unexpected RPC request: ${args.method}`) + }, + }), +}) +const descriptor: ChannelDescriptor = { + authorizedSigner: account.address, + expiringNonceHash: `0x${'11'.repeat(32)}` as Hex, + operator: '0x0000000000000000000000000000000000000000', + payee: '0x742d35cc6634c0532925a3b844bc9e7595f8fe00', + payer: account.address, + salt: `0x${'22'.repeat(32)}` as Hex, + token: '0x20c0000000000000000000000000000000000001', +} +const channelId = Channel.computeId({ + ...descriptor, + chainId: 4217, + escrow: tip20ChannelEscrow, +}) + +function channelEntry(): ChannelEntry { + return { + channelId, + chainId: 4217, + cumulativeAmount: 1n, + deposit: 10n, + descriptor, + escrow: tip20ChannelEscrow, + opened: true, + } +} + +function challengeResponse( + id = 'challenge-1', + snapshot?: SessionSnapshot, + requestOverrides: Record = {}, +): { + challenge: TempoSessionChallenge + response: Response +} { + const challenge = Challenge.from({ + id, + intent: Constants.Intents.session, + method: Constants.Methods.tempo, + realm: 'api.example.test', + request: { + amount: '1', + currency: descriptor.token, + decimals: 0, + methodDetails: { + chainId: 4217, + escrowContract: tip20ChannelEscrow, + sessionProtocol: Constants.SessionProtocols.v2, + ...(snapshot && { [Constants.MethodDetailKeys.sessionSnapshot]: snapshot }), + }, + recipient: descriptor.payee, + unitType: 'request', + ...requestOverrides, + }, + }) as TempoSessionChallenge + return { + challenge, + response: new Response(null, { + status: 402, + headers: { [Constants.Headers.wwwAuthenticate]: Challenge.serialize(challenge) }, + }), + } +} + +function sessionSnapshot(overrides: Partial = {}): SessionSnapshot { + return { + acceptedCumulative: '4', + chainId: 4217, + channelId, + deposit: '10', + descriptor, + escrow: tip20ChannelEscrow, + requiredCumulative: '4', + settled: '0', + spent: '4', + units: 2, + ...overrides, + } +} + +function channelStore(entry = channelEntry()) { + const channels = new Map([[entryKey(entry), entry]]) + const remove = vi.fn((key: string) => { + channels.delete(key) + }) + const store: ChannelStore = { + delete: remove, + get: (key) => channels.get(key), + set(next) { + channels.set(entryKey(next), next) + }, + } + return { remove, store } +} + +function credentialPayload(init: RequestInit | undefined): SessionCredentialPayload | undefined { + const authorization = new Headers(init?.headers).get(Constants.Headers.authorization) + if (!authorization) return undefined + return Credential.deserialize(authorization).payload +} + +function managerParameters(store: ChannelStore) { + return { + account, + channelStore: store, + client, + decimals: 0, + maxDeposit: '10', + } +} + +describe('CLI session manager adapter', () => { + test('rehydrates durable context and closes at receipt-confirmed spend', async () => { + const { challenge } = challengeResponse( + 'challenge-1', + sessionSnapshot({ acceptedCumulative: '3', requiredCumulative: '3', spent: '3' }), + ) + const entry = channelEntry() + entry.cumulativeAmount = 5n + const refreshed = challengeResponse('challenge-2', sessionSnapshot()) + const { remove, store } = channelStore(entry) + const closeUrl = 'https://api.example.test/resource?chainId=testnet' + let closeRequests = 0 + const closeAmounts: string[] = [] + const onChallenge = vi.fn() + const fetch = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { + expect(input.toString()).toBe(closeUrl) + expect(init?.method).toBe('POST') + const payload = credentialPayload(init) + if (payload?.action !== 'close') throw new Error('expected close credential') + expect(payload.channelId).toBe(channelId) + closeAmounts.push(payload.cumulativeAmount) + closeRequests++ + if (closeRequests === 1) return refreshed.response + return new Response(null, { + headers: { + [Constants.Headers.paymentReceipt]: serializeSessionReceipt( + createSessionReceipt({ + acceptedCumulative: 4n, + challengeId: refreshed.challenge.id, + channelId, + spent: 4n, + txHash: `0x${'aa'.repeat(32)}` as Hex, + }), + ), + }, + }) + }) + + const result = await closeWithSessionManager({ + channel: entry, + challenge, + fetch, + input: closeUrl, + manager: managerParameters(store), + onChallenge, + spent: 2n, + }) + + expect(result.receipt).toMatchObject({ channelId, spent: '4' }) + expect(result.manager.state).toMatchObject({ status: 'closed', channelId }) + expect(closeAmounts).toEqual(['3', '4']) + expect(onChallenge).toHaveBeenCalledOnce() + expect(onChallenge).toHaveBeenCalledWith(refreshed.challenge) + expect(fetch).toHaveBeenCalledTimes(2) + expect(remove).toHaveBeenCalledOnce() + }) + + test('rejects refreshed snapshot spend beyond local cumulative authorization', async () => { + const { challenge } = challengeResponse() + const entry = channelEntry() + entry.cumulativeAmount = 5n + const refreshed = challengeResponse( + 'challenge-2', + sessionSnapshot({ acceptedCumulative: '6', requiredCumulative: '6', spent: '6' }), + ) + const { store } = channelStore(entry) + const fetch = vi.fn(async () => refreshed.response) + const onChallenge = vi.fn() + + await expect( + closeWithSessionManager({ + channel: entry, + challenge, + fetch, + input: 'https://api.example.test/resource?chainId=testnet', + manager: managerParameters(store), + onChallenge, + spent: 3n, + }), + ).rejects.toThrow('close snapshot accepted cumulative exceeds local voucher state') + expect(fetch).toHaveBeenCalledOnce() + expect(onChallenge).not.toHaveBeenCalled() + }) + + test('rejects a stored close challenge with a different payee before sending', async () => { + const { challenge } = challengeResponse('challenge-1', undefined, { + recipient: '0x0000000000000000000000000000000000000009', + }) + const entry = channelEntry() + const { store } = channelStore(entry) + const fetch = vi.fn() + + await expect( + closeWithSessionManager({ + channel: entry, + challenge, + fetch, + input: 'https://api.example.test/resource?chainId=testnet', + manager: managerParameters(store), + spent: 1n, + }), + ).rejects.toThrow('Close challenge changed the session payee.') + expect(fetch).not.toHaveBeenCalled() + }) +}) diff --git a/src/cli/sessions/Manager.ts b/src/cli/sessions/Manager.ts new file mode 100644 index 00000000..a408e231 --- /dev/null +++ b/src/cli/sessions/Manager.ts @@ -0,0 +1,93 @@ +import * as Challenge from '../../Challenge.js' +import type { ChannelEntry } from '../../tempo/session/client/ChannelOps.js' +import { resolveEscrow } from '../../tempo/session/client/ChannelOps.js' +import { getSessionManagerInternals } from '../../tempo/session/client/internal/SessionManager.js' +import { sessionManager, type SessionManager } from '../../tempo/session/client/SessionManager.js' +import type { TempoSessionChallenge } from '../../tempo/session/client/Transports.js' +import { + getSessionSnapshot, + isTempoSessionChallenge, +} from '../../tempo/session/client/Transports.js' +import type { SessionReceipt } from '../../tempo/session/precompile/Protocol.js' + +type ManagerParameters = Omit + +/** Inputs for closing a durable session through a newly created manager. */ +export type CloseWithSessionManagerParameters = { + /** Durable open channel entry. */ + channel: ChannelEntry + /** Latest validated challenge for the channel scope. */ + challenge: TempoSessionChallenge + /** Network fetch used for the cooperative close request. Defaults to global fetch. */ + fetch?: typeof globalThis.fetch | undefined + /** Exact resource URL used as the cooperative close endpoint. */ + input: RequestInfo | URL + /** Session manager account, client, policy, and channel-store parameters. */ + manager: ManagerParameters + /** Persists a validated refreshed close challenge before retrying. */ + onChallenge?: ((challenge: TempoSessionChallenge) => void | Promise) | undefined + /** Latest receipt-confirmed spend in raw token units. */ + spent: bigint +} + +/** Result of a rehydrated cooperative close. */ +export type CloseWithSessionManagerResult = { + manager: SessionManager + receipt: SessionReceipt +} + +function resolveFetch(fetch: typeof globalThis.fetch | undefined): typeof globalThis.fetch { + return fetch ?? globalThis.fetch.bind(globalThis) +} + +function assertCloseChallengeScope(challenge: TempoSessionChallenge, channel: ChannelEntry): void { + const chainId = (challenge.request.methodDetails as { chainId?: unknown } | undefined)?.chainId + if (chainId !== undefined && chainId !== channel.chainId) + throw new Error('Close challenge changed the session chain.') + if ( + typeof challenge.request.recipient !== 'string' || + challenge.request.recipient.toLowerCase() !== channel.descriptor.payee.toLowerCase() + ) + throw new Error('Close challenge changed the session payee.') + if ( + typeof challenge.request.currency !== 'string' || + challenge.request.currency.toLowerCase() !== channel.descriptor.token.toLowerCase() + ) + throw new Error('Close challenge changed the session token.') + if (resolveEscrow(challenge).toLowerCase() !== channel.escrow.toLowerCase()) + throw new Error('Close challenge changed the session escrow.') + const snapshot = getSessionSnapshot(challenge) + if (snapshot && snapshot.channelId.toLowerCase() !== channel.channelId.toLowerCase()) + throw new Error('Close challenge changed the session channel.') +} + +/** Rehydrates durable session context and cooperatively closes it through the manager. */ +export async function closeWithSessionManager( + parameters: CloseWithSessionManagerParameters, +): Promise { + assertCloseChallengeScope(parameters.challenge, parameters.channel) + const networkFetch = resolveFetch(parameters.fetch) + let pendingChallenge: TempoSessionChallenge | undefined + const validatedFetch: typeof globalThis.fetch = async (input, init) => { + if (pendingChallenge) { + await parameters.onChallenge?.(pendingChallenge) + pendingChallenge = undefined + } + const response = await networkFetch(input, init) + if (response.status !== 402) return response + const refreshed = Challenge.fromResponseList(response).find(isTempoSessionChallenge) + if (!refreshed) throw new Error('Refreshed close response did not include tempo/session.') + assertCloseChallengeScope(refreshed, parameters.channel) + pendingChallenge = refreshed + return response + } + const manager = sessionManager({ + ...parameters.manager, + bootstrap: false, + fetch: validatedFetch, + }) + getSessionManagerInternals(manager).rehydrate(parameters) + const receipt = await manager.close() + if (!receipt) throw new Error('Session close response did not include a payment receipt.') + return { manager, receipt } +} diff --git a/src/cli/sessions/commands.ts b/src/cli/sessions/commands.ts new file mode 100644 index 00000000..2b399315 --- /dev/null +++ b/src/cli/sessions/commands.ts @@ -0,0 +1,444 @@ +import { Cli, Errors, z } from 'incur' +import { createClient, http } from 'viem' +import { tempo as tempoMainnet, tempoModerato } from 'viem/tempo/chains' + +import { normalizeHeaders } from '../../client/internal/Fetch.js' +import { canSignDescriptor } from '../../tempo/session/client/CredentialState.js' +import * as Chain from '../../tempo/session/precompile/Chain.js' +import * as Channel from '../../tempo/session/precompile/Channel.js' +import { resolvePersistentAccount } from '../account.js' +import { resolveChain, resolveRpcUrl, type Network } from '../utils.js' +import { closeWithSessionManager } from './Manager.js' +import { + createSessionRegistry, + SessionBusyError, + SessionStateError, + type ManagedSession, + type SessionPersistenceContext, + sessionResourceUrl, + sessionScope, + toChannelStore, +} from './store.js' + +type SessionCloseOptions = { + account?: string | undefined + headers?: readonly string[] | undefined + network?: Network | undefined + resourceUrl?: string | undefined + rpcUrl?: string | undefined +} + +const sessionOutputSchema = z.object({ + status: z.enum(['opening', 'open', 'closing', 'stale']), + channelId: z.string(), + account: z.string().optional(), + payer: z.string(), + payee: z.string(), + authorizedSigner: z.string(), + token: z.string(), + escrow: z.string(), + chainId: z.number(), + cumulativeAmount: z.string(), + confirmedSpend: z.string(), + deposit: z.string(), + units: z.number(), + resourceUrl: z.string(), + createdAt: z.string(), + updatedAt: z.string(), +}) + +const sessionCloseOutputSchema = z.object({ + channelId: z.string(), + status: z.enum(['closed', 'already-closed']), + spent: z.string(), + txHash: z.string().optional(), +}) + +const sessionBulkCloseOutputSchema = z.object({ + closed: z.array(sessionCloseOutputSchema), + failed: z.array(z.object({ channelId: z.string(), message: z.string() })), +}) + +type SessionOutput = z.infer +type SessionCloseOutput = z.infer + +function networkChainId(network: Network): number { + return network === 'mainnet' ? tempoMainnet.id : tempoModerato.id +} + +function networkForChain(chainId: number): Network | undefined { + if (chainId === tempoMainnet.id) return 'mainnet' + if (chainId === tempoModerato.id) return 'testnet' + return undefined +} + +function parseHeaders(values: readonly string[] | undefined): Record { + const headers: Record = {} + for (const value of values ?? []) { + const index = value.indexOf(':') + if (index === -1) + throw new Errors.IncurError({ + code: 'INVALID_HEADER', + message: `Invalid header format: ${value}`, + exitCode: 2, + }) + headers[value.slice(0, index).trim()] = value.slice(index + 1).trim() + } + return headers +} + +function outputSession(record: ManagedSession): SessionOutput { + return { + status: record.status, + channelId: record.channel.channelId, + ...(record.account.name && { account: record.account.name }), + payer: record.channel.descriptor.payer, + payee: record.channel.descriptor.payee, + authorizedSigner: record.channel.descriptor.authorizedSigner, + token: record.channel.descriptor.token, + escrow: record.channel.escrow, + chainId: record.channel.chainId, + cumulativeAmount: record.channel.cumulativeAmount.toString(), + confirmedSpend: record.spent.toString(), + deposit: record.channel.deposit.toString(), + units: record.units, + resourceUrl: record.endpoint, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + } +} + +/** Normalizes persistent session failures for CLI output. */ +export function sessionCommandError(error: unknown, fallbackCode: string): never { + if (error instanceof Errors.IncurError) throw error + if (error instanceof SessionBusyError) + throw new Errors.IncurError({ + code: error.code, + message: error.message, + exitCode: error.exitCode, + cause: error, + }) + if (error instanceof SessionStateError) + throw new Errors.IncurError({ + code: error.code, + message: error.message, + exitCode: 65, + cause: error, + }) + throw new Errors.IncurError({ + code: fallbackCode, + message: error instanceof Error ? error.message : String(error), + exitCode: 75, + ...(error instanceof Error && { cause: error }), + }) +} + +async function resolveCloseAccount(record: ManagedSession, accountOverride?: string | undefined) { + const accountName = accountOverride ?? record.account.name + const resolved = await resolvePersistentAccount(accountName) + if (!canSignDescriptor(resolved.account, record.channel.descriptor)) + throw new Errors.IncurError({ + code: 'SESSION_ACCOUNT_MISMATCH', + message: `Account ${resolved.account.address} cannot sign for session ${record.channel.channelId}.`, + exitCode: 2, + }) + return resolved +} + +async function resolveCloseClient( + chainId: number, + options: Pick, +) { + if (options.network && networkChainId(options.network) !== chainId) + throw new Errors.IncurError({ + code: 'CHAIN_MISMATCH', + message: `Session uses chainId ${chainId}, not ${options.network}.`, + exitCode: 2, + }) + const network = options.network ?? networkForChain(chainId) + const rpcUrl = resolveRpcUrl(options.rpcUrl, { network }) + const chain = await resolveChain({ network, rpcUrl }) + if (chain.id !== chainId) + throw new Errors.IncurError({ + code: 'CHAIN_MISMATCH', + message: `Session uses chainId ${chainId}, but RPC is chainId ${chain.id}.`, + exitCode: 2, + }) + return createClient({ chain, transport: http(rpcUrl) }) +} + +const sessions = Cli.create('sessions', { + description: 'Manage persistent payment sessions (list, view, close)', +}) + .command('list', { + description: 'List persistent payment sessions', + options: z.object({ + account: z.string().optional().describe('Filter by account name'), + network: z.enum(['mainnet', 'testnet']).optional().describe('Filter by Tempo network'), + }), + output: z.object({ sessions: z.array(sessionOutputSchema) }), + alias: { account: 'a' }, + async run(c) { + try { + const records = (await createSessionRegistry().list()) + .filter((record) => { + if (c.options.account && record.account.name !== c.options.account) return false + if (c.options.network && record.channel.chainId !== networkChainId(c.options.network)) + return false + return true + }) + .map(outputSession) + const result = { sessions: records } + if (c.format === 'json' && c.formatExplicit) return c.ok(result) + if (records.length === 0) console.log('No sessions.') + for (const [index, record] of records.entries()) { + if (index > 0) console.log('') + printSession(record) + } + return undefined as never + } catch (error) { + return sessionCommandError(error, 'SESSION_LIST_FAILED') + } + }, + }) + .command('view', { + description: 'View a persistent payment session', + args: z.object({ channelId: z.string().describe('Full session channel ID') }), + output: sessionOutputSchema, + async run(c) { + try { + const record = await createSessionRegistry().get(c.args.channelId) + if (!record) + throw new Errors.IncurError({ + code: 'SESSION_NOT_FOUND', + message: `Session ${c.args.channelId} was not found.`, + exitCode: 2, + }) + const result = outputSession(record) + if (c.format === 'json' && c.formatExplicit) return c.ok(result) + printSession(result) + return undefined as never + } catch (error) { + return sessionCommandError(error, 'SESSION_VIEW_FAILED') + } + }, + }) + .command('close', { + description: 'Cooperatively close persistent payment sessions', + usage: [{ suffix: ' [options]' }, { suffix: '--all --yes [options]' }], + args: z.object({ channelId: z.string().optional().describe('Full session channel ID') }), + options: z.object({ + account: z.string().optional().describe('Account name'), + all: z.boolean().optional().default(false).describe('Close every matching session'), + header: z.array(z.string()).optional().describe('Add close request header (repeatable)'), + network: z.enum(['mainnet', 'testnet']).optional().describe('Tempo network'), + rpcUrl: z.string().optional().describe('RPC endpoint (env: MPPX_RPC_URL)'), + url: z.string().optional().describe('Override the stored resource URL'), + yes: z.boolean().optional().default(false).describe('Confirm closing every session'), + }), + output: z.union([sessionCloseOutputSchema, sessionBulkCloseOutputSchema]), + alias: { account: 'a', header: 'H', rpcUrl: 'r' }, + async run(c) { + if (c.options.all && c.args.channelId) + return c.error({ + code: 'INVALID_SESSION_CLOSE', + message: 'Specify a channel ID or --all, not both.', + exitCode: 2, + }) + if (!c.options.all && !c.args.channelId) + return c.error({ + code: 'INVALID_SESSION_CLOSE', + message: 'Specify a channel ID or --all.', + exitCode: 2, + }) + if (c.options.all && !c.options.yes) + return c.error({ + code: 'CONFIRMATION_REQUIRED', + message: 'Closing all sessions requires --yes.', + exitCode: 2, + }) + if (c.options.all && (c.options.header || c.options.rpcUrl || c.options.url)) + return c.error({ + code: 'INVALID_SESSION_CLOSE', + message: '--header, --rpc-url, and --url only apply to a single session.', + exitCode: 2, + }) + + const registry = createSessionRegistry() + async function closeSession( + channelId: string, + options: SessionCloseOptions = {}, + ): Promise { + const candidate = await registry.get(channelId) + if (!candidate) + throw new Errors.IncurError({ + code: 'SESSION_NOT_FOUND', + message: `Session ${channelId} was not found.`, + exitCode: 2, + }) + const scope = sessionScope(candidate.channel) + const lock = await registry.acquire(scope) + try { + const record = await registry.get(channelId) + if (!record) + throw new Errors.IncurError({ + code: 'SESSION_NOT_FOUND', + message: `Session ${channelId} was not found.`, + exitCode: 2, + }) + const resolvedAccount = await resolveCloseAccount(record, options.account) + const client = await resolveCloseClient(record.channel.chainId, options) + const expectedId = Channel.computeId({ + ...record.channel.descriptor, + escrow: record.channel.escrow, + chainId: record.channel.chainId, + }) + if (expectedId.toLowerCase() !== record.channel.channelId.toLowerCase()) + throw new Errors.IncurError({ + code: 'SESSION_STATE_INVALID', + message: 'Stored descriptor does not derive the session channel ID.', + exitCode: 65, + }) + + const state = await Chain.getChannelState( + client as never, + record.channel.channelId, + record.channel.escrow, + ) + if (state.deposit === 0n) { + await registry.remove(record.channel.channelId) + return { + channelId: record.channel.channelId, + status: 'already-closed', + spent: record.spent.toString(), + } + } + + const endpoint = sessionResourceUrl(options.resourceUrl ?? record.endpoint) + const account = { + ...(resolvedAccount.source === 'keychain' && { name: resolvedAccount.accountName }), + address: resolvedAccount.account.address, + } + const closingContext = (challenge = record.challenge): SessionPersistenceContext => ({ + status: 'closing', + account, + endpoint, + challenge, + ...(record.receipt && { receipt: record.receipt }), + spent: record.spent, + units: record.units, + }) + const closing = await registry.upsert({ + ...closingContext(), + channel: record.channel, + }) + const headers = parseHeaders(options.headers) + let latestChallenge = closing.challenge + const closeFetch: typeof globalThis.fetch = async (input, init) => + globalThis.fetch(input, { + ...init, + headers: { ...headers, ...normalizeHeaders(init?.headers) }, + }) + const result = await closeWithSessionManager({ + channel: closing.channel, + challenge: closing.challenge, + fetch: closeFetch, + input: endpoint, + spent: closing.spent, + async onChallenge(challenge) { + latestChallenge = challenge + await registry.upsert({ + ...closingContext(challenge), + channel: closing.channel, + }) + }, + manager: { + account: resolvedAccount.account, + client, + channelStore: toChannelStore(registry, { + scope, + selection: closing.channel.channelId, + context: () => closingContext(latestChallenge), + }), + }, + }) + await registry.remove(closing.channel.channelId) + return { + channelId: closing.channel.channelId, + status: 'closed', + spent: result.receipt.spent, + txHash: result.receipt.txHash, + } + } finally { + await lock.release() + } + } + + try { + if (c.options.all) { + const records = (await registry.list()).filter( + (record) => + (!c.options.account || record.account.name === c.options.account) && + (!c.options.network || record.channel.chainId === networkChainId(c.options.network)), + ) + const closed: SessionCloseOutput[] = [] + const failed: { channelId: string; message: string }[] = [] + for (const record of records) { + try { + closed.push( + await closeSession(record.channel.channelId, { + account: c.options.account, + network: c.options.network, + }), + ) + } catch (error) { + failed.push({ + channelId: record.channel.channelId, + message: error instanceof Error ? error.message : String(error), + }) + } + } + const result = { closed, failed } + if (failed.length > 0) { + for (const failure of failed) + process.stderr.write(`${failure.channelId}: ${failure.message}\n`) + process.exitCode = 1 + } + if (c.format === 'json' && c.formatExplicit) return c.ok(result) + for (const item of closed) console.log(`${item.channelId} ${item.status} ${item.spent}`) + for (const failure of failed) + console.log(`${failure.channelId} failed ${failure.message}`) + return undefined as never + } + + const result = await closeSession(c.args.channelId!, { + account: c.options.account, + headers: c.options.header, + network: c.options.network, + resourceUrl: c.options.url, + rpcUrl: c.options.rpcUrl, + }) + if (c.format === 'json' && c.formatExplicit) return c.ok(result) + console.log(`${result.channelId} ${result.status} ${result.spent}`) + if (result.txHash) console.log(` transaction ${result.txHash}`) + return undefined as never + } catch (error) { + return sessionCommandError(error, 'SESSION_CLOSE_FAILED') + } + }, + }) + +function printSession(record: SessionOutput): void { + console.log(record.channelId) + console.log(` status ${record.status}`) + console.log(` cumulative amount ${record.cumulativeAmount}`) + console.log(` confirmed spend ${record.confirmedSpend}`) + console.log(` deposit ${record.deposit}`) + console.log(` chain ${record.chainId}`) + console.log(` payer ${record.payer}`) + console.log(` payee ${record.payee}`) + console.log(` token ${record.token}`) + console.log(` resource ${record.resourceUrl}`) +} + +export default sessions diff --git a/src/cli/sessions/request.test.ts b/src/cli/sessions/request.test.ts new file mode 100644 index 00000000..a243554f --- /dev/null +++ b/src/cli/sessions/request.test.ts @@ -0,0 +1,51 @@ +import type { Hex } from 'viem' +import { describe, expect, test } from 'vp/test' + +import type * as Challenge from '../../Challenge.js' +import { resolveSessionMaxDeposit, resolveSessionSelection } from './request.js' + +const channelId = `0x${'12'.repeat(32)}` as Hex +describe('resolveSessionSelection', () => { + test('uses auto by default and accepts new or an explicit channel', () => { + expect(resolveSessionSelection('auto', undefined)).toBe('auto') + expect(resolveSessionSelection('new', undefined)).toBe('new') + expect(resolveSessionSelection(channelId.toUpperCase().replace('0X', '0x'), undefined)).toBe( + channelId, + ) + }) + + test('supports the channel method compatibility alias', () => { + expect(resolveSessionSelection('auto', channelId)).toBe(channelId) + expect(resolveSessionSelection(channelId, channelId)).toBe(channelId) + }) + + test('rejects conflicting selectors', () => { + expect(() => resolveSessionSelection('new', channelId)).toThrow( + '--session and -M channel= select different sessions.', + ) + }) +}) + +describe('resolveSessionMaxDeposit', () => { + const challenge = { + id: 'challenge-1', + realm: 'api.example.test', + method: 'tempo', + intent: 'session', + request: { + amount: '1000000', + currency: '0x3333333333333333333333333333333333333333', + decimals: 6, + recipient: '0x2222222222222222222222222222222222222222', + suggestedDeposit: '7000000', + }, + } satisfies Challenge.Challenge + + test('converts the raw server suggestion to human-readable token units', () => { + expect(resolveSessionMaxDeposit(challenge, {}, false)).toBe('7') + }) + + test('prefers the human-readable CLI deposit override', () => { + expect(resolveSessionMaxDeposit(challenge, { deposit: '10' }, false)).toBe('10') + }) +}) diff --git a/src/cli/sessions/request.ts b/src/cli/sessions/request.ts new file mode 100644 index 00000000..402ca17b --- /dev/null +++ b/src/cli/sessions/request.ts @@ -0,0 +1,353 @@ +import { Errors } from 'incur' +import type { Hex } from 'ox' +import { createClient, formatUnits, http } from 'viem' + +import type * as Challenge from '../../Challenge.js' +import { + canSignDescriptor, + resolveChallengeContext, +} from '../../tempo/session/client/CredentialState.js' +import { getSessionManagerInternals } from '../../tempo/session/client/internal/SessionManager.js' +import { sessionManager } from '../../tempo/session/client/SessionManager.js' +import type { TempoSessionChallenge } from '../../tempo/session/client/Transports.js' +import { isEventStream, type SessionReceipt } from '../../tempo/session/precompile/Protocol.js' +import { resolvePersistentAccount } from '../account.js' +import { + isTestnet, + printResponseHeaders, + resolveChain, + resolveRpcUrl, + type Network, +} from '../utils.js' +import { + createSessionRegistry, + type ManagedSession, + type SessionPersistenceContext, + type SessionRegistry, + type SessionSelection, + type SessionStatus, + SessionBusyError, + sessionResourceUrl, + sessionScope, + sessionScopeKey, + toChannelStore, +} from './store.js' + +/** CLI options needed to run a persistent Tempo session request. */ +export type PersistentSessionRequestOptions = { + account?: string | undefined + fail?: boolean | undefined + include?: boolean | undefined + network?: Network | undefined + rpcUrl?: string | undefined + session: string + silent: boolean + verbose: number +} + +/** Inputs for a persistent request after challenge selection and confirmation. */ +export type PersistentSessionRequestParameters = { + challenge: TempoSessionChallenge + challengeResponse: Response + endpoint: string + fetch: typeof globalThis.fetch + fetchInput: RequestInfo | URL + init: RequestInit + info(message: string): void + methodOptions: Record + options: PersistentSessionRequestOptions + registry?: SessionRegistry | undefined +} + +/** Resolves `--session` and the `-M channel=` compatibility alias. */ +export function resolveSessionSelection( + session: string, + channelAlias: string | undefined, +): SessionSelection { + if (channelAlias && session !== 'auto' && channelAlias.toLowerCase() !== session.toLowerCase()) + throw new Errors.IncurError({ + code: 'SESSION_SELECTION_CONFLICT', + message: '--session and -M channel= select different sessions.', + exitCode: 2, + }) + const value = channelAlias ?? session + if (value === 'auto' || value === 'new') return value + if (/^0x[0-9a-fA-F]{64}$/.test(value)) return value.toLowerCase() as Hex.Hex + throw new Errors.IncurError({ + code: 'INVALID_SESSION', + message: 'Session must be auto, new, or a 32-byte channel ID.', + exitCode: 2, + }) +} + +function sessionDecimals(challenge: Challenge.Challenge): number { + return typeof challenge.request.decimals === 'number' ? challenge.request.decimals : 6 +} + +/** @internal Resolves the manager deposit cap in human-readable token units. */ +export function resolveSessionMaxDeposit( + challenge: Challenge.Challenge, + methodOptions: Record, + testnet: boolean, +): string | undefined { + if (methodOptions.deposit !== undefined) return methodOptions.deposit + const suggested = challenge.request.suggestedDeposit + if (typeof suggested === 'string') + return formatUnits(BigInt(suggested), sessionDecimals(challenge)) + return testnet ? '10' : undefined +} + +function validateReceipt( + receipt: SessionReceipt | null | undefined, + parameters: { + challenge: Challenge.Challenge + channelId: string + cumulative: bigint + }, +): SessionReceipt | undefined { + if (!receipt) return undefined + if (receipt.challengeId !== parameters.challenge.id) + throw new Error('Session receipt challenge does not match the paid request.') + if (receipt.channelId.toLowerCase() !== parameters.channelId.toLowerCase()) + throw new Error('Session receipt channel does not match the selected channel.') + const accepted = BigInt(receipt.acceptedCumulative) + const spent = BigInt(receipt.spent) + if (accepted > parameters.cumulative) + throw new Error('Session receipt exceeds the local cumulative authorization.') + if (spent > accepted) throw new Error('Session receipt spend exceeds its accepted authorization.') + return receipt +} + +function sameChannel(record: ManagedSession, channelId: string): boolean { + return record.channel.channelId.toLowerCase() === channelId.toLowerCase() +} + +function writeSseChunk(chunk: string): void { + if (chunk.trim() === '[DONE]') return + if (chunk.length === 0) { + process.stdout.write('\n') + return + } + try { + const parsed = JSON.parse(chunk) as { + token?: string + choices?: { delta?: { content?: string } }[] + } + process.stdout.write(parsed.token ?? parsed.choices?.[0]?.delta?.content ?? chunk) + } catch { + process.stdout.write(chunk) + } +} + +/** Runs one manager-backed request while holding the payer and payment-scope lock. */ +export async function runPersistentSessionRequest( + parameters: PersistentSessionRequestParameters, +): Promise { + const { options } = parameters + const resolvedAccount = await resolvePersistentAccount(options.account) + const rpcUrl = resolveRpcUrl(options.rpcUrl, { network: options.network }) + const chain = await resolveChain({ network: options.network, rpcUrl }) + const client = createClient({ chain, transport: http(rpcUrl) }) + const challengeContext = await resolveChallengeContext({ + challenge: parameters.challenge, + getClient: async () => client, + }) + if (challengeContext.chainId !== chain.id) + throw new Errors.IncurError({ + code: 'CHAIN_MISMATCH', + message: `Challenge requires chainId ${challengeContext.chainId}, but RPC is chainId ${chain.id}.`, + exitCode: 2, + }) + + const scope = { + payer: resolvedAccount.account.address, + payee: challengeContext.payee, + token: challengeContext.token, + escrow: challengeContext.escrow, + chainId: challengeContext.chainId, + } + const selection = resolveSessionSelection(options.session, parameters.methodOptions.channel) + const registry = parameters.registry ?? createSessionRegistry() + const lock = await registry.acquire(scope).catch((cause: unknown) => { + if (cause instanceof SessionBusyError) + throw new Errors.IncurError({ + code: cause.code, + message: cause.message, + exitCode: cause.exitCode, + cause, + }) + throw cause + }) + + try { + const selectedId = + selection === 'auto' + ? await registry.getPreferred(scope) + : selection === 'new' + ? undefined + : selection + const selected = selectedId ? await registry.get(selectedId) : undefined + if (selectedId && !selected) + throw new Errors.IncurError({ + code: 'SESSION_NOT_FOUND', + message: `Session ${selectedId} was not found.`, + exitCode: 2, + }) + if (selected && selection !== 'auto') { + if ( + selected.status !== 'open' || + !selected.channel.opened || + sessionScopeKey(sessionScope(selected.channel)) !== sessionScopeKey(scope) || + !canSignDescriptor(resolvedAccount.account, selected.channel.descriptor) + ) + throw new Errors.IncurError({ + code: 'SESSION_MISMATCH', + message: `Session ${selected.channel.channelId} cannot be used for this account and challenge.`, + exitCode: 2, + }) + } + + const reusable = selected?.status === 'open' && selected.channel.opened ? selected : undefined + let status: SessionStatus = reusable ? 'open' : 'opening' + let latestChallenge = parameters.challenge + let latestReceipt = reusable?.receipt + let spent = reusable?.spent ?? 0n + let units = reusable?.units ?? 0 + const endpoint = sessionResourceUrl(parameters.endpoint) + const account = { + ...(resolvedAccount.source === 'keychain' && { name: resolvedAccount.accountName }), + address: resolvedAccount.account.address, + } + const persistenceContext = (): SessionPersistenceContext => ({ + status, + account, + endpoint, + challenge: latestChallenge, + ...(latestReceipt && { receipt: latestReceipt }), + spent, + units, + }) + const channelStore = toChannelStore(registry, { + scope, + selection, + context: persistenceContext, + onNewChannel() { + status = 'opening' + latestReceipt = undefined + spent = 0n + units = 0 + }, + }) + const { signal, ...baseInit } = parameters.init + const requestInit = { + ...baseInit, + ...(signal && { signal }), + onReceipt(receipt: SessionReceipt) { + latestReceipt = receipt + spent = spent > BigInt(receipt.spent) ? spent : BigInt(receipt.spent) + units = Math.max(units, receipt.units ?? 0) + }, + } + if (parameters.challengeResponse.status !== 402 || parameters.challengeResponse.bodyUsed) + throw new Error('Session manager requires an unconsumed 402 challenge response.') + let replayPending = true + const manager = sessionManager({ + account: resolvedAccount.account, + bootstrap: false, + client, + channelStore, + decimals: sessionDecimals(parameters.challenge), + maxDeposit: resolveSessionMaxDeposit( + parameters.challenge, + parameters.methodOptions, + isTestnet(chain), + ), + fetch: async (input, init) => { + if (!replayPending) return parameters.fetch(input, init) + replayPending = false + return parameters.challengeResponse + }, + }) + if (reusable) + getSessionManagerInternals(manager).rehydrate({ + channel: reusable.channel, + challenge: parameters.challenge, + input: parameters.fetchInput, + spent: reusable.spent, + }) + const { onReceipt, ...managerInit } = requestInit + const response = await manager.fetch(parameters.fetchInput, managerInit) + + if (options.fail && response.status >= 400) + throw new Errors.IncurError({ + code: 'HTTP_ERROR', + message: `HTTP error ${response.status}`, + exitCode: 22, + }) + if (response.status === 402) + throw new Errors.IncurError({ + code: 'PAYMENT_REJECTED', + message: 'Payment rejected.', + exitCode: 75, + }) + + printResponseHeaders(response, { + include: false, + verbose: options.include ? Math.max(options.verbose, 2) : options.verbose, + silent: options.silent, + }) + latestChallenge = response.challenge ?? parameters.challenge + const channelId = manager.channelId + if (!channelId) throw new Error('Session manager did not select a channel.') + latestReceipt = + validateReceipt(response.receipt, { + challenge: latestChallenge, + channelId, + cumulative: manager.cumulative, + }) ?? latestReceipt + + if (isEventStream(response)) { + const stream = getSessionManagerInternals(manager).consumeSseResponse( + parameters.fetchInput, + response, + { onReceipt, signal: parameters.init.signal ?? undefined }, + ) + for await (const chunk of stream) writeSseChunk(chunk) + } else { + process.stdout.write(Buffer.from(await response.arrayBuffer())) + } + + const record = await registry.get(channelId) + if (!record || !sameChannel(record, channelId)) + throw new Error(`Session ${channelId} was not persisted.`) + if (latestReceipt) { + const validatedReceipt = validateReceipt(latestReceipt, { + challenge: latestChallenge, + channelId, + cumulative: manager.cumulative, + }) + if (!validatedReceipt) throw new Error('Session receipt validation failed.') + latestReceipt = validatedReceipt + spent = spent > BigInt(validatedReceipt.spent) ? spent : BigInt(validatedReceipt.spent) + units = Math.max(units, validatedReceipt.units ?? 0) + status = 'open' + } + await registry.upsert({ + status, + channel: record.channel, + account, + endpoint, + challenge: latestChallenge, + ...(latestReceipt && { receipt: latestReceipt }), + spent, + units, + }) + + if (!options.silent) { + parameters.info(`Session retained ${channelId}\n`) + parameters.info(`Close with: mppx sessions close ${channelId}\n`) + } + } finally { + await lock.release() + } +} diff --git a/src/cli/sessions/store.test.ts b/src/cli/sessions/store.test.ts new file mode 100644 index 00000000..05d2b34d --- /dev/null +++ b/src/cli/sessions/store.test.ts @@ -0,0 +1,581 @@ +import * as fs from 'node:fs/promises' +import * as os from 'node:os' +import * as path from 'node:path' + +import type { Address, Hex } from 'viem' + +vi.mock('node:fs/promises', async (importOriginal) => ({ + ...(await importOriginal()), +})) + +import type * as Challenge from '../../Challenge.js' +import type { ChannelEntry } from '../../tempo/session/client/ChannelOps.js' +import { entryKey } from '../../tempo/session/client/ChannelStore.js' +import type { SessionReceipt } from '../../tempo/session/precompile/Protocol.js' +import sessions from './commands.js' +import { + createSessionRegistry, + SessionBusyError, + SessionStateError, + sessionScopeKey, + toChannelStore, + type SessionPersistenceContext, + type SessionRegistry, + type SessionScope, +} from './store.js' + +const payer = '0x1111111111111111111111111111111111111111' as Address +const payee = '0x2222222222222222222222222222222222222222' as Address +const token = '0x3333333333333333333333333333333333333333' as Address +const escrow = '0x4444444444444444444444444444444444444444' as Address +const operator = '0x0000000000000000000000000000000000000000' as Address +const channelId = `0x${'aa'.repeat(32)}` as Hex +const mainnetChannelId = `0x${'bb'.repeat(32)}` as Hex + +let temporaryDirectory: string +let stateRoot: string + +beforeEach(async () => { + temporaryDirectory = await fs.mkdtemp(path.join(os.tmpdir(), 'mppx-sessions-')) + vi.stubEnv('XDG_STATE_HOME', temporaryDirectory) + stateRoot = path.join(temporaryDirectory, 'mppx', 'sessions', 'v1') +}) + +afterEach(async () => { + vi.unstubAllEnvs() + await fs.rm(temporaryDirectory, { force: true, recursive: true }) +}) + +function channel(overrides: Partial = {}): ChannelEntry { + return { + channelId, + cumulativeAmount: 10n, + deposit: 100n, + descriptor: { + payer, + payee, + operator, + token, + salt: `0x${'55'.repeat(32)}`, + authorizedSigner: payer, + expiringNonceHash: `0x${'66'.repeat(32)}`, + }, + escrow, + chainId: 42431, + opened: true, + ...overrides, + } +} + +function challenge(id = 'challenge-1', chainId = 42431): Challenge.Challenge { + return { + id, + realm: 'api.example.test', + method: 'tempo', + intent: 'session', + request: { + amount: '1', + currency: token, + recipient: payee, + methodDetails: { chainId, escrowContract: escrow }, + }, + } +} + +function receipt(overrides: Partial = {}): SessionReceipt { + return { + method: 'tempo', + intent: 'session', + status: 'success', + timestamp: '2026-07-16T00:01:00.000Z', + reference: channelId, + challengeId: 'challenge-1', + channelId, + acceptedCumulative: '10', + spent: '2', + units: 3, + ...overrides, + } +} + +function scope(overrides: Partial = {}): SessionScope { + return { payer, payee, token, escrow, chainId: 42431, ...overrides } +} + +function registryOptions( + overrides: Parameters[0] = {}, +): Parameters[0] { + return { stateRoot, ...overrides } +} + +async function serveSessions(argv: string[]) { + let output = '' + let exitCode: number | undefined + await sessions.serve(argv, { + stdout(value: string) { + output += value + }, + exit(code: number) { + exitCode = code + }, + }) + return { exitCode, output } +} + +async function seedCommandSessions(registry: SessionRegistry) { + await registry.upsert({ + status: 'open', + channel: channel({ + cumulativeAmount: 9_007_199_254_740_993_123_456_789n, + deposit: 9_999_999_999_999_999_999_999_999n, + }), + account: { name: 'testnet-payer', address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet&sql=select%201', + challenge: challenge('testnet-challenge'), + spent: 1_234_567_890_123_456_789n, + units: 7, + }) + await registry.upsert({ + status: 'closing', + channel: channel({ channelId: mainnetChannelId, chainId: 4217, cumulativeAmount: 20n }), + account: { name: 'mainnet-payer', address: payer }, + endpoint: 'https://api.example.test/query?chainId=mainnet', + challenge: challenge('mainnet-challenge', 4217), + spent: 5n, + units: 2, + }) +} + +function commandRegistry() { + const timestamps = [new Date('2026-07-16T00:00:00.000Z'), new Date('2026-07-16T00:01:00.000Z')] + return createSessionRegistry(registryOptions({ now: () => timestamps.shift()! })) +} + +describe('createSessionRegistry', () => { + test('uses XDG_STATE_HOME for the versioned state root', () => { + const registry = createSessionRegistry() + expect(registry.root).toBe(stateRoot) + }) + + test('persists sanitized state atomically with private permissions', async () => { + const times = [new Date('2026-07-16T00:00:00.000Z'), new Date('2026-07-15T00:00:00.000Z')] + const registry = createSessionRegistry( + registryOptions({ now: () => times.shift() ?? new Date('2026-07-17T00:00:00.000Z') }), + ) + const firstChannel = Object.assign(channel(), { privateKey: 'do-not-store' }) + const firstAccount = Object.assign( + { name: 'testnet', address: payer }, + { privateKey: 'do-not-store' }, + ) + + const first = await registry.upsert({ + status: 'opening', + channel: firstChannel, + account: firstAccount, + endpoint: 'https://api.example.test/query?chainId=testnet&sql=select%201#response', + challenge: challenge(), + receipt: receipt(), + spent: 2n, + units: 3, + }) + const second = await registry.upsert({ + status: 'open', + channel: channel({ cumulativeAmount: 8n }), + account: { address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet&sql=select%201#ignored', + challenge: challenge('challenge-2'), + spent: 1n, + units: 1, + }) + + expect(first.endpoint).toBe('https://api.example.test/query?chainId=testnet&sql=select%201') + expect(second).toMatchObject({ + status: 'open', + account: { name: 'testnet', address: payer }, + spent: 2n, + units: 3, + createdAt: '2026-07-16T00:00:00.000Z', + updatedAt: '2026-07-16T00:00:00.000Z', + }) + expect(second.channel.cumulativeAmount).toBe(10n) + expect(second.receipt).toEqual(receipt()) + + const file = path.join(stateRoot, 'channels', `${channelId}.json`) + const source = await fs.readFile(file, 'utf8') + const stored: unknown = JSON.parse(source) + expect(stored).toMatchObject({ + version: 1, + method: 'tempo', + intent: 'session', + channel: { cumulativeAmount: '10', deposit: '100' }, + spent: '2', + units: 3, + }) + expect(source).not.toContain('privateKey') + expect((await fs.stat(stateRoot)).mode & 0o777).toBe(0o700) + expect((await fs.stat(path.join(stateRoot, 'channels'))).mode & 0o777).toBe(0o700) + expect((await fs.stat(file)).mode & 0o777).toBe(0o600) + expect( + (await fs.readdir(path.join(stateRoot, 'channels'))).filter((name) => name.includes('.tmp-')), + ).toEqual([]) + }) + + test('manages preferred sessions and removes mappings before records', async () => { + const registry = createSessionRegistry(registryOptions()) + await registry.upsert({ + status: 'open', + channel: channel(), + account: { name: 'testnet', address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet', + challenge: challenge(), + }) + + await registry.setPreferred(scope(), channelId) + expect(await registry.getPreferred(scope())).toBe(channelId) + await registry.clearPreferred(scope(), `0x${'bb'.repeat(32)}`) + expect(await registry.getPreferred(scope())).toBe(channelId) + expect(await registry.list()).toEqual([ + expect.objectContaining({ channel: expect.objectContaining({ channelId }) }), + ]) + await expect( + registry.setPreferred( + scope({ payee: '0x7777777777777777777777777777777777777777' }), + channelId, + ), + ).rejects.toBeInstanceOf(SessionStateError) + + await registry.remove(channelId) + expect(await registry.get(channelId)).toBeUndefined() + expect(await registry.getPreferred(scope())).toBeUndefined() + }) + + test('does not overwrite a corrupt managed record', async () => { + const file = path.join(stateRoot, 'channels', `${channelId}.json`) + await fs.mkdir(path.dirname(file), { recursive: true }) + await fs.writeFile(file, '{invalid json') + + await expect(createSessionRegistry(registryOptions()).get(channelId)).rejects.toMatchObject({ + code: 'SESSION_STATE_INVALID', + file, + }) + await expect( + createSessionRegistry(registryOptions()).upsert({ + status: 'open', + channel: channel(), + account: { address: payer }, + endpoint: 'https://api.example.test/query', + challenge: challenge(), + }), + ).rejects.toBeInstanceOf(SessionStateError) + expect(await fs.readFile(file, 'utf8')).toBe('{invalid json') + }) + + test('rejects live and remote locks, then reclaims a dead same-host lock', async () => { + const first = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 101, isProcessAlive: () => true }), + ) + const live = await first.acquire(scope()) + const contender = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 202, isProcessAlive: (pid) => pid === 101 }), + ) + + await expect(contender.acquire(scope())).rejects.toMatchObject({ + code: 'SESSION_BUSY', + exitCode: 75, + scope: sessionScopeKey(scope()), + owner: { hostname: 'host-a', pid: 101 }, + }) + await live.release() + + const deadOwner = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 303, isProcessAlive: () => true }), + ) + const abandoned = await deadOwner.acquire(scope()) + const reclaimer = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 404, isProcessAlive: () => false }), + ) + const reclaimed = await reclaimer.acquire(scope()) + await abandoned.release() + const observer = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 405, isProcessAlive: () => true }), + ) + await expect(observer.acquire(scope())).rejects.toBeInstanceOf(SessionBusyError) + await reclaimed.release() + + const remoteOwner = createSessionRegistry( + registryOptions({ hostname: 'host-b', pid: 505, isProcessAlive: () => true }), + ) + const remote = await remoteOwner.acquire(scope()) + await expect(reclaimer.acquire(scope())).rejects.toMatchObject({ + code: 'SESSION_BUSY', + owner: { hostname: 'host-b', pid: 505 }, + }) + await remote.release() + }) + + test('allows only one concurrent dead-lock reclaimer to acquire', async () => { + const deadOwner = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 101, isProcessAlive: () => true }), + ) + await deadOwner.acquire(scope()) + const [lockName] = await fs.readdir(path.join(stateRoot, 'locks')) + if (!lockName) throw new Error('Expected a lock file.') + const lock = path.join(stateRoot, 'locks', lockName) + const stale = JSON.parse(await fs.readFile(lock, 'utf8')) as { token: string } + + let unsafeUnlinks = 0 + let releaseFirstUnlink: (() => void) | undefined + const secondUnlink = new Promise((resolve) => { + releaseFirstUnlink = resolve + }) + const { unlink } = await vi.importActual('node:fs/promises') + const unlinkSpy = vi.spyOn(fs, 'unlink').mockImplementation(async (file) => { + if (String(file) !== lock) return unlink(file) + try { + await fs.access(`${lock}.reclaim`) + return unlink(file) + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error + } + unsafeUnlinks++ + if (unsafeUnlinks === 1) { + await secondUnlink + return unlink(file) + } + if (unsafeUnlinks === 2) { + releaseFirstUnlink?.() + await vi.waitUntil( + async () => { + try { + const current = JSON.parse(await fs.readFile(lock, 'utf8')) as { token: string } + return current.token !== stale.token + } catch { + return false + } + }, + { interval: 1, timeout: 1_000 }, + ) + } + return unlink(file) + }) + + const isProcessAlive = (pid: number) => pid === 202 || pid === 303 + const first = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 202, isProcessAlive }), + ) + const second = createSessionRegistry( + registryOptions({ hostname: 'host-a', pid: 303, isProcessAlive }), + ) + try { + const results = await Promise.allSettled([first.acquire(scope()), second.acquire(scope())]) + unlinkSpy.mockRestore() + const acquired = results.filter( + (result): result is PromiseFulfilledResult>> => + result.status === 'fulfilled', + ) + const rejected = results.filter( + (result): result is PromiseRejectedResult => result.status === 'rejected', + ) + expect(acquired).toHaveLength(1) + expect(rejected).toHaveLength(1) + expect(rejected[0]?.reason).toBeInstanceOf(SessionBusyError) + await acquired[0]?.value.release() + } finally { + unlinkSpy.mockRestore() + } + }) +}) + +describe('toChannelStore', () => { + test('uses preferred open sessions and never reuses opening sessions', async () => { + const registry = createSessionRegistry(registryOptions()) + let status: SessionPersistenceContext['status'] = 'opening' + const context = (): SessionPersistenceContext => ({ + status, + account: { name: 'testnet', address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet', + challenge: challenge(), + }) + const store = toChannelStore(registry, { scope: scope(), selection: 'new', context }) + + expect(await store.get(entryKey(channel()))).toBeUndefined() + await store.set(channel()) + expect((await registry.get(channelId))?.status).toBe('opening') + expect(await registry.getPreferred(scope())).toBe(channelId) + expect(await store.get(entryKey(channel()))).toBeUndefined() + + await store.delete(entryKey(channel())) + expect((await registry.get(channelId))?.status).toBe('stale') + expect(await registry.getPreferred(scope())).toBeUndefined() + + status = 'open' + const reopened = toChannelStore(registry, { scope: scope(), selection: 'new', context }) + await reopened.set(channel()) + const automatic = toChannelStore(registry, { scope: scope(), selection: 'auto', context }) + expect(await automatic.get(entryKey(channel()))).toEqual(channel()) + }) + + test('retains closing state when the manager deletes after creating a close credential', async () => { + const registry = createSessionRegistry(registryOptions()) + let status: SessionPersistenceContext['status'] = 'open' + const context = (): SessionPersistenceContext => ({ + status, + account: { address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet', + challenge: challenge(), + spent: 4n, + units: 5, + }) + const store = toChannelStore(registry, { scope: scope(), selection: 'new', context }) + await store.set(channel()) + status = 'closing' + + await store.delete(entryKey(channel())) + + expect(await registry.get(channelId)).toMatchObject({ + status: 'closing', + spent: 4n, + units: 5, + }) + expect(await registry.getPreferred(scope())).toBe(channelId) + }) + + test('resets persistence context when a rejected channel is replaced', async () => { + const registry = createSessionRegistry(registryOptions()) + await registry.upsert({ + status: 'open', + channel: channel(), + account: { address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet', + challenge: challenge(), + receipt: receipt(), + spent: 2n, + units: 3, + }) + await registry.setPreferred(scope(), channelId) + + let status: SessionPersistenceContext['status'] = 'open' + let latestReceipt: SessionReceipt | undefined = receipt() + let spent = 2n + let units = 3 + const context = (): SessionPersistenceContext => ({ + status, + account: { address: payer }, + endpoint: 'https://api.example.test/query?chainId=testnet', + challenge: challenge(), + ...(latestReceipt && { receipt: latestReceipt }), + spent, + units, + }) + const store = toChannelStore(registry, { + scope: scope(), + selection: 'auto', + context, + onNewChannel() { + status = 'opening' + latestReceipt = undefined + spent = 0n + units = 0 + }, + }) + expect(await store.get(entryKey(channel()))).toEqual(channel()) + await store.delete(entryKey(channel())) + + const replacementId = `0x${'bb'.repeat(32)}` as Hex + await store.set(channel({ channelId: replacementId })) + + expect(await registry.get(channelId)).toMatchObject({ status: 'stale', spent: 2n, units: 3 }) + expect(await registry.get(replacementId)).toMatchObject({ + status: 'opening', + spent: 0n, + units: 0, + }) + expect((await registry.get(replacementId))?.receipt).toBeUndefined() + }) +}) + +describe('session commands', () => { + test('list and view return stable JSON projections with filters', async () => { + await seedCommandSessions(commandRegistry()) + + const listed = JSON.parse((await serveSessions(['list', '--json'])).output) + expect(listed.sessions).toEqual([ + { + status: 'open', + channelId, + account: 'testnet-payer', + payer, + payee, + authorizedSigner: payer, + token, + escrow, + chainId: 42431, + cumulativeAmount: '9007199254740993123456789', + confirmedSpend: '1234567890123456789', + deposit: '9999999999999999999999999', + units: 7, + resourceUrl: 'https://api.example.test/query?chainId=testnet&sql=select%201', + createdAt: '2026-07-16T00:00:00.000Z', + updatedAt: '2026-07-16T00:00:00.000Z', + }, + expect.objectContaining({ + status: 'closing', + channelId: mainnetChannelId, + account: 'mainnet-payer', + chainId: 4217, + cumulativeAmount: '20', + confirmedSpend: '5', + }), + ]) + + const viewed = await serveSessions(['view', channelId, '--json']) + const byAccount = await serveSessions(['list', '--account', 'testnet-payer', '--json']) + const byNetwork = await serveSessions(['list', '--network', 'mainnet', '--json']) + const mismatch = await serveSessions([ + 'list', + '--account', + 'testnet-payer', + '--network', + 'mainnet', + '--json', + ]) + expect(JSON.parse(viewed.output)).toEqual(listed.sessions[0]) + expect(JSON.parse(byAccount.output).sessions).toMatchObject([{ channelId }]) + expect(JSON.parse(byNetwork.output).sessions).toMatchObject([{ channelId: mainnetChannelId }]) + expect(JSON.parse(mismatch.output)).toEqual({ sessions: [] }) + }) + + test('view rejects a missing channel', async () => { + const missing = `0x${'cc'.repeat(32)}` + const result = await serveSessions(['view', missing, '--json']) + expect(result).toMatchObject({ exitCode: 2 }) + expect(result.output).toContain(`Session ${missing} was not found.`) + }) + + test('close all reports failures in session order', async () => { + await seedCommandSessions(commandRegistry()) + vi.stubEnv('MPPX_PRIVATE_KEY', `0x${'11'.repeat(32)}`) + const originalExitCode = process.exitCode + const originalWrite = process.stderr.write + let stderr = '' + process.stderr.write = ((chunk: unknown) => { + stderr += String(chunk) + return true + }) as typeof process.stderr.write + try { + const result = await serveSessions(['close', '--all', '--yes', '--json']) + const output = JSON.parse(result.output) + expect(output.closed).toEqual([]) + expect(output.failed.map((failure: { channelId: string }) => failure.channelId)).toEqual([ + channelId, + mainnetChannelId, + ]) + expect(process.exitCode).toBe(1) + expect(stderr).toContain(channelId) + expect(stderr).toContain(mainnetChannelId) + } finally { + process.exitCode = originalExitCode + process.stderr.write = originalWrite + } + }) +}) diff --git a/src/cli/sessions/store.ts b/src/cli/sessions/store.ts new file mode 100644 index 00000000..8419569b --- /dev/null +++ b/src/cli/sessions/store.ts @@ -0,0 +1,940 @@ +import { createHash, randomUUID } from 'node:crypto' +import * as fs from 'node:fs/promises' +import * as os from 'node:os' +import * as path from 'node:path' + +import type { Address, Hex } from 'viem' + +import * as Challenge from '../../Challenge.js' +import { resolveEscrow, type ChannelEntry } from '../../tempo/session/client/ChannelOps.js' +import { + deserializeEntry, + entryKey, + serializeEntry, + type ChannelStore, + type StoredChannel, +} from '../../tempo/session/client/ChannelStore.js' +import { + isTempoSessionChallenge, + type TempoSessionChallenge, +} from '../../tempo/session/client/Transports.js' +import type { SessionReceipt } from '../../tempo/session/precompile/Protocol.js' +import * as z from '../../zod.js' + +const sessionStateVersion = 1 as const + +/** Lifecycle state recorded for a managed CLI session. */ +export type SessionStatus = 'opening' | 'open' | 'closing' | 'stale' + +/** Account identity required to resume or close a managed session. */ +export type SessionAccount = { + /** Optional mppx account name. */ + name?: string | undefined + /** Payer wallet address. */ + address: Address +} + +/** Payment scope used to isolate preferred sessions and process locks. */ +export type SessionScope = { + payer: Address + payee: Address + token: Address + escrow: Address + chainId: number +} + +/** Durable session record returned by the CLI registry. */ +export type ManagedSession = { + version: typeof sessionStateVersion + status: SessionStatus + channel: ChannelEntry + account: SessionAccount + endpoint: string + challenge: TempoSessionChallenge + receipt?: SessionReceipt | undefined + spent: bigint + units: number + createdAt: string + updatedAt: string +} + +/** Input persisted by {@link SessionRegistry.upsert}. */ +export type SessionUpsert = { + status: SessionStatus + channel: ChannelEntry + account: SessionAccount + endpoint: string + challenge: Challenge.Challenge + receipt?: SessionReceipt | undefined + spent?: bigint | undefined + units?: number | undefined +} + +/** Dynamic context used when adapting the registry to the SDK channel store. */ +export type SessionPersistenceContext = Omit + +/** Selection policy used by a persistent CLI request. */ +export type SessionSelection = 'auto' | 'new' | Hex + +/** Held process lock for a session scope. */ +export type SessionLock = { + /** Releases the lock if this process still owns it. */ + release(): Promise +} + +/** Filesystem-backed persistent session registry. */ +export type SessionRegistry = { + /** Versioned registry root. */ + readonly root: string + /** Returns a managed session by full channel ID. */ + get(channelId: string): Promise + /** Lists managed sessions. */ + list(): Promise + /** Creates or monotonically updates a managed session. */ + upsert(input: SessionUpsert): Promise + /** Removes a validated managed session and its preferred mappings. */ + remove(channelId: string): Promise + /** Returns the preferred channel ID for a payer and payment scope. */ + getPreferred(scope: SessionScope): Promise + /** Sets the preferred channel after verifying it matches the scope. */ + setPreferred(scope: SessionScope, channelId: string): Promise + /** Clears the preferred channel, optionally only when it matches `channelId`. */ + clearPreferred(scope: SessionScope, channelId?: string | undefined): Promise + /** Acquires an exclusive process lock for a payer and payment scope. */ + acquire(scope: SessionScope): Promise +} + +/** Options for {@link createSessionRegistry}. */ +export type CreateSessionRegistryOptions = { + /** Override the versioned state root. */ + stateRoot?: string | undefined + /** Host identity written to lock files. */ + hostname?: string | undefined + /** Process ID written to lock files. */ + pid?: number | undefined + /** Clock used for persisted timestamps. */ + now?: (() => Date) | undefined + /** Process liveness probe used for same-host lock reclamation. */ + isProcessAlive?: ((pid: number) => boolean) | undefined +} + +/** Invalid, corrupt, or inconsistent persistent session state. */ +export class SessionStateError extends Error { + override readonly name = 'SessionStateError' + readonly code = 'SESSION_STATE_INVALID' + readonly file?: string | undefined + + constructor(message: string, options: { cause?: unknown; file?: string | undefined } = {}) { + super(message, options.cause === undefined ? undefined : { cause: options.cause }) + this.file = options.file + } +} + +/** A session scope currently owned by another live process. */ +export class SessionBusyError extends Error { + override readonly name = 'SessionBusyError' + readonly code = 'SESSION_BUSY' + readonly exitCode = 75 + readonly scope: string + readonly owner: { hostname: string; pid: number } + + constructor(scope: string, owner: { hostname: string; pid: number }) { + super(`Session scope is busy in process ${owner.pid} on ${owner.hostname}.`) + this.scope = scope + this.owner = owner + } +} + +const storedChannelSchema = z.object({ + channelId: z.hash(), + cumulativeAmount: z.string(), + deposit: z.string(), + descriptor: z.object({ + payer: z.address(), + payee: z.address(), + operator: z.address(), + token: z.address(), + salt: z.hash(), + authorizedSigner: z.address(), + expiringNonceHash: z.hash(), + }), + escrow: z.address(), + chainId: z.number(), + opened: z.boolean(), +}) +const accountSchema = z.object({ + name: z.optional(z.string()), + address: z.address(), +}) +const receiptSchema = z.object({ + method: z.literal('tempo'), + intent: z.literal('session'), + status: z.literal('success'), + timestamp: z.string(), + reference: z.string(), + challengeId: z.string(), + channelId: z.hash(), + acceptedCumulative: z.string(), + spent: z.string(), + units: z.optional(z.number()), + txHash: z.optional(z.hash()), +}) +const storedSessionSchema = z.object({ + version: z.literal(sessionStateVersion), + method: z.literal('tempo'), + intent: z.literal('session'), + status: z.enum(['opening', 'open', 'closing', 'stale']), + channel: storedChannelSchema, + account: accountSchema, + endpoint: z.string(), + challenge: Challenge.Schema, + receipt: z.optional(receiptSchema), + spent: z.string(), + units: z.number(), + createdAt: z.string(), + updatedAt: z.string(), +}) +const preferredSessionSchema = z.object({ + version: z.literal(sessionStateVersion), + channelId: z.hash(), +}) +const lockOwnerSchema = z.object({ + version: z.literal(sessionStateVersion), + scope: z.string(), + hostname: z.string(), + pid: z.number(), + token: z.string(), + createdAt: z.string(), +}) + +type StoredManagedSession = Omit< + z.infer, + 'account' | 'challenge' | 'channel' | 'receipt' +> & { + account: SessionAccount + challenge: TempoSessionChallenge + channel: StoredChannel + receipt?: SessionReceipt | undefined +} +type PreferredSession = z.infer +type LockOwner = z.infer + +function parseStored( + schema: schema, + value: unknown, + file: string | undefined, + message: string, +): z.output { + const parsed = schema.safeParse(value) + if (!parsed.success) throw stateError(file, message, parsed.error) + return parsed.data +} + +type RegistryPaths = { + root: string + channels: string + locks: string + preferred: string +} + +/** Returns the stable payer-qualified key for a persistent session scope. */ +export function sessionScopeKey(scope: SessionScope): string { + const normalized = normalizeScope(scope) + return [ + normalized.payer, + normalized.payee, + normalized.token, + normalized.escrow, + normalized.chainId, + ].join(':') +} + +/** Returns the persistent payment scope for a channel. */ +export function sessionScope(channel: ChannelEntry): SessionScope { + return { + payer: channel.descriptor.payer, + payee: channel.descriptor.payee, + token: channel.descriptor.token, + escrow: channel.escrow, + chainId: channel.chainId, + } +} + +/** Creates a filesystem-backed CLI session registry. */ +export function createSessionRegistry(options: CreateSessionRegistryOptions = {}): SessionRegistry { + const root = options.stateRoot ?? sessionStateRoot() + const paths: RegistryPaths = { + root, + channels: path.join(root, 'channels'), + locks: path.join(root, 'locks'), + preferred: path.join(root, 'preferred'), + } + const hostname = options.hostname ?? os.hostname() + const pid = options.pid ?? process.pid + const now = options.now ?? (() => new Date()) + const isProcessAlive = options.isProcessAlive ?? processIsAlive + + async function ensureDirectories(): Promise { + for (const directory of [paths.root, paths.channels, paths.locks, paths.preferred]) { + await fs.mkdir(directory, { recursive: true, mode: 0o700 }) + await fs.chmod(directory, 0o700) + } + } + + async function get(channelId: string): Promise { + const normalizedId = normalizeChannelId(channelId) + const file = channelFile(paths, normalizedId) + const value = await readJson(file) + if (value === undefined) return undefined + const record = parseStoredSession(value, file) + if (record.channel.channelId.toLowerCase() !== normalizedId) + throw stateError(file, 'Session filename does not match its channel ID.') + return deserializeSession(record) + } + + async function list(): Promise { + let entries: import('node:fs').Dirent[] + try { + entries = await fs.readdir(paths.channels, { withFileTypes: true }) + } catch (error) { + if (hasCode(error, 'ENOENT')) return [] + throw stateError(paths.channels, 'Unable to list managed sessions.', error) + } + + const records: ManagedSession[] = [] + for (const entry of entries) { + if (!entry.isFile() || entry.name.includes('.tmp-')) continue + if (!entry.name.endsWith('.json')) continue + const channelId = entry.name.slice(0, -'.json'.length) + normalizeChannelId(channelId) + const record = await get(channelId) + if (record) records.push(record) + } + return records.sort((left, right) => left.createdAt.localeCompare(right.createdAt)) + } + + async function upsert(input: SessionUpsert): Promise { + await ensureDirectories() + const channel = sanitizeChannel(input.channel) + const normalizedId = normalizeChannelId(channel.channelId) + const file = channelFile(paths, normalizedId) + const previousValue = await readJson(file) + const previous = + previousValue === undefined ? undefined : parseStoredSession(previousValue, file) + if (previous) assertSameSession(previous, input, file) + + const challenge = parseSessionChallenge(input.challenge, file) + assertChallengeMatchesChannel(challenge, channel, file) + const receipt = input.receipt + ? sanitizeReceipt(input.receipt, channel.channelId, file) + : undefined + const previousChannel = previous ? deserializeEntry(previous.channel) : undefined + const cumulativeAmount = maxBigInt( + channel.cumulativeAmount, + previousChannel?.cumulativeAmount ?? 0n, + ) + const spent = maxBigInt( + input.spent ?? 0n, + receipt ? BigInt(receipt.spent) : 0n, + previous ? BigInt(previous.spent) : 0n, + ) + if (spent > cumulativeAmount) + throw stateError(file, 'Session spend exceeds the locally authorized cumulative amount.') + + const storedChannel = sanitizeStoredChannel( + serializeEntry({ ...channel, cumulativeAmount }), + file, + ) + const timestamp = monotonicTimestamp(previous?.updatedAt, now(), file) + const latestReceipt = selectLatestReceipt(previous?.receipt, receipt) + const account = sanitizeAccount(input.account, file) + const record: StoredManagedSession = { + version: sessionStateVersion, + method: 'tempo', + intent: 'session', + status: input.status, + channel: storedChannel, + account: { + ...(account.name !== undefined + ? { name: account.name } + : previous?.account.name !== undefined + ? { name: previous.account.name } + : {}), + address: account.address, + }, + endpoint: sessionResourceUrl(input.endpoint, file), + challenge, + ...(latestReceipt !== undefined && { receipt: latestReceipt }), + spent: spent.toString(), + units: Math.max(previous?.units ?? 0, input.units ?? 0, receipt?.units ?? 0), + createdAt: previous?.createdAt ?? timestamp, + updatedAt: timestamp, + } + const parsed = parseStoredSession(record, file) + await writeJsonAtomic(file, parsed) + return deserializeSession(parsed) + } + + async function getPreferred(scope: SessionScope): Promise { + const file = preferredFile(paths, scope) + const value = await readJson(file) + if (value === undefined) return undefined + const { channelId } = parseStored( + preferredSessionSchema, + value, + file, + 'Preferred session is invalid.', + ) + const record = await get(channelId) + if (!record) throw stateError(file, `Preferred session ${channelId} does not exist.`) + assertChannelScope(record.channel, scope, file) + return channelId as Hex + } + + async function setPreferred(scope: SessionScope, channelId: string): Promise { + const normalizedId = normalizeChannelId(channelId) + const record = await get(normalizedId) + const file = preferredFile(paths, scope) + if (!record) throw stateError(file, `Session ${normalizedId} does not exist.`) + assertChannelScope(record.channel, scope, file) + await writeJsonAtomic(file, { + version: sessionStateVersion, + channelId: normalizedId as Hex, + } satisfies PreferredSession) + } + + async function clearPreferred( + scope: SessionScope, + channelId?: string | undefined, + ): Promise { + const file = preferredFile(paths, scope) + const normalizedId = channelId === undefined ? undefined : normalizeChannelId(channelId) + const value = await readJson(file) + if (value === undefined) return + const current = parseStored( + preferredSessionSchema, + value, + file, + 'Preferred session is invalid.', + ).channelId + if (normalizedId && current.toLowerCase() !== normalizedId) return + await removeFile(file, 'Unable to clear preferred session.') + } + + async function remove(channelId: string): Promise { + const normalizedId = normalizeChannelId(channelId) + const record = await get(normalizedId) + if (!record) return + await clearPreferred(sessionScope(record.channel), normalizedId) + const file = channelFile(paths, normalizedId) + await removeFile(file, 'Unable to remove session.') + } + + async function acquireKey(scope: string): Promise { + await ensureDirectories() + const file = lockFile(paths, scope) + for (let attempt = 0; attempt < 3; attempt++) { + const owner: LockOwner = { + version: sessionStateVersion, + scope, + hostname, + pid, + token: randomUUID(), + createdAt: now().toISOString(), + } + try { + await createLock(file, owner) + if (await fileExists(deadLockClaimFile(file))) { + await removeOwnedLock(file, owner) + continue + } + return { + async release() { + await removeOwnedLock(file, owner) + }, + } + } catch (error) { + if (!hasCode(error, 'EEXIST')) throw error + } + + const value = await readJson(file) + if (value === undefined) continue + const current = parseStored(lockOwnerSchema, value, file, 'Session lock is invalid.') + if (current.scope !== scope) throw stateError(file, 'Session lock scope is invalid.') + if (current.hostname !== hostname || isProcessAlive(current.pid)) + throw new SessionBusyError(scope, current) + await removeDeadLock(file, current) + } + const value = await readJson(file) + if (value === undefined) throw stateError(file, 'Unable to acquire session lock.') + const owner = parseStored(lockOwnerSchema, value, file, 'Session lock is invalid.') + throw new SessionBusyError(scope, owner) + } + + async function acquire(scope: SessionScope): Promise { + return acquireKey(sessionScopeKey(scope)) + } + + return { + root, + get, + list, + upsert, + remove, + getPreferred, + setPreferred, + clearPreferred, + acquire, + } +} + +/** Adapts a persistent registry selection to the session manager's channel store. */ +export function toChannelStore( + registry: SessionRegistry, + options: { + scope: SessionScope + selection: SessionSelection + context: () => SessionPersistenceContext + onNewChannel?: ((channel: ChannelEntry) => void) | undefined + }, +): ChannelStore { + const expectedKey = scopeEntryKey(options.scope) + let selectedChannelId: Hex | undefined = + options.selection === 'auto' || options.selection === 'new' + ? undefined + : (normalizeChannelId(options.selection) as Hex) + + function assertKey(key: string): void { + if (key.toLowerCase() !== expectedKey) + throw new SessionStateError('Session manager requested an unexpected payment scope.') + } + + async function selected(reusableOnly = true): Promise { + if (options.selection === 'new' && !selectedChannelId) return undefined + const channelId = selectedChannelId ?? (await registry.getPreferred(options.scope)) + if (!channelId) return undefined + const record = await registry.get(channelId) + if (!record) throw new SessionStateError(`Session ${channelId} does not exist.`) + assertChannelScope(record.channel, options.scope) + if (reusableOnly && (record.status !== 'open' || !record.channel.opened)) return undefined + selectedChannelId = channelId + return record + } + + return { + async get(key) { + assertKey(key) + return (await selected())?.channel + }, + async set(channel) { + assertKey(entryKey(channel)) + assertChannelScope(channel, options.scope) + if (selectedChannelId && selectedChannelId.toLowerCase() !== channel.channelId.toLowerCase()) + throw new SessionStateError( + `Session manager selected ${selectedChannelId}, but attempted to store ${channel.channelId}.`, + ) + if (!selectedChannelId) options.onNewChannel?.(channel) + const record = await registry.upsert({ ...options.context(), channel }) + selectedChannelId = record.channel.channelId + await registry.setPreferred(options.scope, record.channel.channelId) + }, + async delete(key) { + assertKey(key) + const record = await selected(false) + if (!record) return + const context = options.context() + await registry.upsert({ + ...record, + ...context, + status: context.status === 'closing' ? 'closing' : 'stale', + channel: record.channel, + spent: context.spent ?? record.spent, + units: context.units ?? record.units, + }) + if (context.status !== 'closing') + await registry.clearPreferred(options.scope, record.channel.channelId) + selectedChannelId = undefined + }, + } +} + +function sessionStateRoot(): string { + const stateHome = process.env.XDG_STATE_HOME || path.join(os.homedir(), '.local', 'state') + return path.join(stateHome, 'mppx', 'sessions', `v${sessionStateVersion}`) +} + +function channelFile(paths: RegistryPaths, channelId: string): string { + return path.join(paths.channels, `${channelId}.json`) +} + +function lockFile(paths: RegistryPaths, scope: string): string { + const digest = createHash('sha256').update(scope).digest('hex') + return path.join(paths.locks, `${digest}.lock`) +} + +function preferredFile(paths: RegistryPaths, scope: SessionScope): string { + const digest = createHash('sha256').update(sessionScopeKey(scope)).digest('hex') + return path.join(paths.preferred, `${digest}.json`) +} + +function normalizeChannelId(channelId: string): string { + return parseStored( + z.hash(), + channelId, + undefined, + `Invalid session channel ID: ${channelId}.`, + ).toLowerCase() +} + +function normalizeAddress(value: unknown, label: string, file?: string | undefined): Address { + return parseStored(z.address(), value, file, `Invalid ${label}.`).toLowerCase() as Address +} + +function normalizeScope(scope: SessionScope): SessionScope { + if (!Number.isSafeInteger(scope.chainId) || scope.chainId < 0) + throw new SessionStateError('Invalid session chain ID.') + return { + payer: normalizeAddress(scope.payer, 'session payer'), + payee: normalizeAddress(scope.payee, 'session payee'), + token: normalizeAddress(scope.token, 'session token'), + escrow: normalizeAddress(scope.escrow, 'session escrow'), + chainId: scope.chainId, + } +} + +function scopeEntryKey(scope: SessionScope): string { + const normalized = normalizeScope(scope) + return [normalized.payee, normalized.token, normalized.escrow, normalized.chainId].join(':') +} + +/** Returns the exact HTTP resource URL persisted for session management. */ +export function sessionResourceUrl(endpoint: unknown, file?: string | undefined): string { + if (typeof endpoint !== 'string') throw stateError(file, 'Session endpoint is invalid.') + let parsed: URL + try { + parsed = new URL(endpoint) + } catch (cause) { + throw stateError(file, 'Session endpoint is invalid.', cause) + } + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') + throw stateError(file, 'Session endpoint must use HTTP or HTTPS.') + if (parsed.username || parsed.password) + throw stateError(file, 'Session endpoint must not contain credentials.') + const fragment = endpoint.indexOf('#') + return fragment === -1 ? endpoint : endpoint.slice(0, fragment) +} + +function sanitizeAccount(value: unknown, file?: string | undefined): SessionAccount { + return parseStored(accountSchema, value, file, 'Session account is invalid.') as SessionAccount +} + +function sanitizeChannel(channel: ChannelEntry): ChannelEntry { + const stored = sanitizeStoredChannel(serializeEntry(channel)) + return deserializeEntry(stored) +} + +function sanitizeStoredChannel(value: unknown, file?: string | undefined): StoredChannel { + return parseStored( + storedChannelSchema, + value, + file, + 'Stored session channel is invalid.', + ) as StoredChannel +} + +function parseSessionChallenge(value: unknown, file?: string | undefined): TempoSessionChallenge { + const parsed = Challenge.Schema.safeParse(value) + if (!parsed.success || !isTempoSessionChallenge(parsed.data)) + throw stateError(file, 'Stored session challenge is invalid.') + return parsed.data +} + +function sanitizeReceipt( + value: unknown, + channelId: string, + file?: string | undefined, +): SessionReceipt { + const receipt = parseStored( + receiptSchema, + value, + file, + 'Stored session receipt is invalid.', + ) as SessionReceipt + if (receipt.channelId !== channelId.toLowerCase()) + throw stateError(file, 'Stored session receipt has a different channel ID.') + if (receipt.reference.toLowerCase() !== channelId.toLowerCase()) + throw stateError(file, 'Stored session receipt has a different reference.') + if (BigInt(receipt.spent) > BigInt(receipt.acceptedCumulative)) + throw stateError(file, 'Stored session receipt spend exceeds its accepted amount.') + return receipt +} + +function parseStoredSession(value: unknown, file: string): StoredManagedSession { + const candidate = parseStored(storedSessionSchema, value, file, 'Stored session is invalid.') + const { receipt: candidateReceipt, ...rest } = candidate + const channel = candidate.channel as StoredChannel + const account = candidate.account as SessionAccount + if (account.address.toLowerCase() !== channel.descriptor.payer.toLowerCase()) + throw stateError(file, 'Stored session account does not match the channel payer.') + const receipt = candidateReceipt + ? sanitizeReceipt(candidateReceipt, channel.channelId, file) + : undefined + const spent = candidate.spent + const cumulativeAmount = BigInt(channel.cumulativeAmount) + if (BigInt(spent) > cumulativeAmount) + throw stateError(file, 'Stored session spend exceeds its cumulative authorization.') + if (receipt && BigInt(receipt.acceptedCumulative) > cumulativeAmount) + throw stateError(file, 'Stored receipt exceeds the cumulative authorization.') + if (Date.parse(candidate.updatedAt) < Date.parse(candidate.createdAt)) + throw stateError(file, 'Stored session timestamps are not monotonic.') + const endpoint = sessionResourceUrl(candidate.endpoint, file) + if (endpoint !== candidate.endpoint) + throw stateError(file, 'Stored session endpoint contains a fragment.') + const challenge = parseSessionChallenge(candidate.challenge, file) + assertChallengeMatchesChannel(challenge, deserializeEntry(channel), file) + return { + ...rest, + account, + channel, + endpoint, + challenge, + ...(receipt && { receipt }), + } +} + +function deserializeSession(record: StoredManagedSession): ManagedSession { + return { + version: sessionStateVersion, + status: record.status, + channel: deserializeEntry(record.channel), + account: record.account, + endpoint: record.endpoint, + challenge: record.challenge, + ...(record.receipt && { receipt: record.receipt }), + spent: BigInt(record.spent), + units: record.units, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + } +} + +function assertSameSession( + previous: StoredManagedSession, + input: SessionUpsert, + file: string, +): void { + const channel = sanitizeChannel(input.channel) + const account = sanitizeAccount(input.account, file) + const previousChannel = deserializeEntry(previous.channel) + if (JSON.stringify(channelIdentity(channel)) !== JSON.stringify(channelIdentity(previousChannel))) + throw stateError(file, 'Session update changed immutable channel identity.') + if (account.address.toLowerCase() !== previous.account.address.toLowerCase()) + throw stateError(file, 'Session update changed the payer account.') +} + +function channelIdentity(channel: ChannelEntry): object { + return { + channelId: channel.channelId.toLowerCase(), + descriptor: Object.fromEntries( + Object.entries(channel.descriptor).map(([key, value]) => [ + key, + typeof value === 'string' ? value.toLowerCase() : value, + ]), + ), + escrow: channel.escrow.toLowerCase(), + chainId: channel.chainId, + } +} + +function assertChallengeMatchesChannel( + challenge: Challenge.Challenge, + channel: ChannelEntry, + file?: string | undefined, +): void { + const payee = normalizeAddress(challenge.request.recipient, 'challenge recipient', file) + const token = normalizeAddress(challenge.request.currency, 'challenge currency', file) + if (payee !== channel.descriptor.payee.toLowerCase()) + throw stateError(file, 'Session challenge payee does not match the channel.') + if (token !== channel.descriptor.token.toLowerCase()) + throw stateError(file, 'Session challenge token does not match the channel.') + if (resolveEscrow(challenge).toLowerCase() !== channel.escrow.toLowerCase()) + throw stateError(file, 'Session challenge escrow does not match the channel.') + if (isObject(challenge.request.methodDetails)) { + const methodDetails = challenge.request.methodDetails + if (methodDetails.chainId !== undefined && methodDetails.chainId !== channel.chainId) + throw stateError(file, 'Session challenge chain does not match the channel.') + } +} + +function assertChannelScope( + channel: ChannelEntry, + scope: SessionScope, + file?: string | undefined, +): void { + const normalized = normalizeScope(scope) + if ( + channel.descriptor.payer.toLowerCase() !== normalized.payer || + channel.descriptor.payee.toLowerCase() !== normalized.payee || + channel.descriptor.token.toLowerCase() !== normalized.token || + channel.escrow.toLowerCase() !== normalized.escrow || + channel.chainId !== normalized.chainId + ) + throw stateError(file, 'Session channel does not match the selected payment scope.') +} + +function selectLatestReceipt( + previous: SessionReceipt | undefined, + next: SessionReceipt | undefined, +): SessionReceipt | undefined { + if (!previous) return next + if (!next) return previous + return Date.parse(next.timestamp) >= Date.parse(previous.timestamp) ? next : previous +} + +function monotonicTimestamp(previous: string | undefined, next: Date, file: string): string { + if (!Number.isFinite(next.getTime())) throw stateError(file, 'Session clock is invalid.') + const nextTimestamp = next.toISOString() + if (!previous || Date.parse(nextTimestamp) >= Date.parse(previous)) return nextTimestamp + return previous +} + +function maxBigInt(...values: bigint[]): bigint { + return values.reduce((maximum, value) => (value > maximum ? value : maximum), 0n) +} + +async function readJson(file: string): Promise { + let source: string + try { + source = await fs.readFile(file, 'utf8') + } catch (error) { + if (hasCode(error, 'ENOENT')) return undefined + throw stateError(file, 'Unable to read session state.', error) + } + try { + const value: unknown = JSON.parse(source) + return value + } catch (cause) { + throw stateError(file, 'Session state contains invalid JSON.', cause) + } +} + +async function writeJsonAtomic(file: string, value: unknown): Promise { + await fs.mkdir(path.dirname(file), { recursive: true, mode: 0o700 }) + await fs.chmod(path.dirname(file), 0o700) + const temporary = `${file}.tmp-${process.pid}-${randomUUID()}` + let handle: fs.FileHandle | undefined + try { + handle = await fs.open(temporary, 'wx', 0o600) + await handle.writeFile(`${JSON.stringify(value, null, 2)}\n`, 'utf8') + await handle.sync() + await handle.close() + handle = undefined + await fs.rename(temporary, file) + await fs.chmod(file, 0o600) + await syncDirectory(path.dirname(file)) + } catch (cause) { + await handle?.close().catch(() => undefined) + await fs.unlink(temporary).catch(() => undefined) + if (cause instanceof SessionStateError) throw cause + throw stateError(file, 'Unable to write session state.', cause) + } +} + +async function createLock(file: string, owner: LockOwner): Promise { + let handle: fs.FileHandle | undefined + try { + handle = await fs.open(file, 'wx', 0o600) + await handle.writeFile(`${JSON.stringify(owner)}\n`, 'utf8') + await handle.sync() + await handle.close() + handle = undefined + await syncDirectory(path.dirname(file)) + } catch (error) { + await handle?.close().catch(() => undefined) + if (handle) await fs.unlink(file).catch(() => undefined) + throw error + } +} + +async function removeDeadLock(file: string, expected: LockOwner): Promise { + const claim = deadLockClaimFile(file) + try { + await fs.link(file, claim) + } catch (error) { + if (hasCode(error, 'EEXIST') || hasCode(error, 'ENOENT')) return + throw stateError(file, 'Unable to claim dead lock.', error) + } + + try { + const currentValue = await readJson(claim) + if (currentValue === undefined) return + const current = parseStored(lockOwnerSchema, currentValue, claim, 'Session lock is invalid.') + if (current.token !== expected.token) return + await removeFile(file, 'Unable to reclaim dead lock.') + } finally { + await removeFile(claim, 'Unable to release dead lock claim.') + } +} + +function deadLockClaimFile(file: string): string { + return `${file}.reclaim` +} + +async function fileExists(file: string): Promise { + try { + await fs.access(file) + return true + } catch (error) { + if (hasCode(error, 'ENOENT')) return false + throw stateError(file, 'Unable to inspect session lock.', error) + } +} + +async function removeOwnedLock(file: string, expected: LockOwner): Promise { + const currentValue = await readJson(file) + if (currentValue === undefined) return + const current = parseStored(lockOwnerSchema, currentValue, file, 'Session lock is invalid.') + if (current.token !== expected.token) return + await removeFile(file, 'Unable to release session lock.') +} + +async function removeFile(file: string, message: string): Promise { + try { + await fs.unlink(file) + await syncDirectory(path.dirname(file)) + } catch (error) { + if (!hasCode(error, 'ENOENT')) throw stateError(file, message, error) + } +} + +async function syncDirectory(directory: string): Promise { + const handle = await fs.open(directory, 'r') + try { + await handle.sync() + } finally { + await handle.close() + } +} + +function processIsAlive(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch (error) { + if (hasCode(error, 'ESRCH')) return false + if (hasCode(error, 'EPERM')) return true + throw error + } +} + +function isObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +function hasCode(error: unknown, code: string): boolean { + return error instanceof Error && 'code' in error && error.code === code +} + +function stateError(file: string | undefined, message: string, cause?: unknown): SessionStateError { + return new SessionStateError(message, { ...(cause !== undefined && { cause }), file }) +} diff --git a/src/tempo/session/client/ChannelOps.test.ts b/src/tempo/session/client/ChannelOps.test.ts index 6209c766..49a393d8 100644 --- a/src/tempo/session/client/ChannelOps.test.ts +++ b/src/tempo/session/client/ChannelOps.test.ts @@ -9,6 +9,17 @@ import * as Types from '../precompile/Protocol.js' import * as Voucher from '../precompile/Voucher.js' import * as ChannelOps from './ChannelOps.js' +const mocks = vi.hoisted(() => ({ + prepareTransactionRequest: vi.fn(async (_client: unknown, request: unknown) => request), + signTransaction: vi.fn(async () => '0x1234'), +})) + +vi.mock('viem/actions', async (importOriginal) => ({ + ...(await importOriginal()), + prepareTransactionRequest: mocks.prepareTransactionRequest, + signTransaction: mocks.signTransaction, +})) + const account = privateKeyToAccount( '0xac0974bec39a17e36ba6a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80', ) @@ -160,4 +171,33 @@ describe('precompile client ChannelOps credential builders', () => { ), ).toBe(true) }) + + test('distinguishes otherwise-identical fee-sponsored management transactions', async () => { + mocks.prepareTransactionRequest.mockClear() + let randomValue = 1 + const random = vi.spyOn(globalThis.crypto, 'getRandomValues').mockImplementation((array) => { + if (!array) return array + new Uint8Array(array.buffer, array.byteOffset, array.byteLength).fill(randomValue++) + return array + }) + try { + await ChannelOps.createTopUpPayload(client, account, descriptor, 10n, chainId, true) + await ChannelOps.createTopUpPayload(client, account, descriptor, 10n, chainId, true) + } finally { + random.mockRestore() + } + + const requests = mocks.prepareTransactionRequest.mock.calls.map(([, request]) => request) + const validAfter = requests.map((request) => + Number((request as { validAfter?: number }).validAfter), + ) + const now = Math.floor(Date.now() / 1_000) + + expect(requests).toMatchObject([ + { feePayer: true, validAfter: expect.any(Number) }, + { feePayer: true, validAfter: expect.any(Number) }, + ]) + expect(new Set(validAfter).size).toBe(2) + expect(validAfter.every((value) => value >= 0 && value < now)).toBe(true) + }) }) diff --git a/src/tempo/session/client/ChannelOps.ts b/src/tempo/session/client/ChannelOps.ts index be08def8..b590d7b4 100644 --- a/src/tempo/session/client/ChannelOps.ts +++ b/src/tempo/session/client/ChannelOps.ts @@ -67,6 +67,14 @@ function readAccessKeyAddress(account: Account): Address | undefined { return readOptionalAddress((account as AccountWithAccessKey).accessKeyAddress) } +/** Returns random past seconds to distinguish otherwise-identical expiring transactions. */ +function randomValidAfter(): number { + const now = BigInt(Math.floor(Date.now() / 1_000)) + const latest = now - 60n + if (latest <= 0n) return 0 + return Number(BigInt(Hex.random(8)) % latest) +} + /** Resolves the voucher authority address for a client account. */ export function resolveAuthorizedSigner(account: Account): Address { return readAccessKeyAddress(account) ?? account.address @@ -79,16 +87,17 @@ async function prepareTempoChannelTransaction( calls: readonly TempoChannelCall[] feePayer?: boolean | undefined feeToken: Address + validAfter?: number | undefined }, ) { - const { account, calls, feePayer, feeToken } = parameters + const { account, calls, feePayer, feeToken, validAfter } = parameters // viem's stable transaction request type does not yet expose Tempo's // `calls`, `feePayer`, and `feeToken` fields together. Keep the cast at // this boundary so session credential builders stay typed. return prepareTransactionRequest(client, { account, calls, - ...(feePayer ? { feePayer: true } : {}), + ...(feePayer ? { feePayer: true, ...(validAfter !== undefined ? { validAfter } : {}) } : {}), feeToken, } as never) } @@ -316,6 +325,7 @@ export async function createTopUpPayload( ], feePayer, feeToken: descriptor.token, + ...(feePayer ? { validAfter: randomValidAfter() } : {}), }) const transaction = await signPreparedTempoTransaction(client, prepared) diff --git a/src/tempo/session/client/SessionManager.test.ts b/src/tempo/session/client/SessionManager.test.ts index 0e29d699..15a77f3d 100644 --- a/src/tempo/session/client/SessionManager.test.ts +++ b/src/tempo/session/client/SessionManager.test.ts @@ -1082,7 +1082,7 @@ describe('Session', () => { expect(voucherPayload.channelId).toBe(openPayload.channelId) expect(voucherPayload.descriptor).toEqual(openPayload.descriptor) expect(voucherPayload.cumulativeAmount).toBe('2000000') - expect(requestedUrls[2]).toBe('https://api.example.com/stream') + expect(requestedUrls[2]).toBe('https://api.example.com/stream?prompt=paid') }) test('ignores precompile SSE voucher requests for a different channel', async () => { @@ -1321,6 +1321,7 @@ describe('Session', () => { challengeId, channelId: payload.channelId, spent: BigInt(payload.cumulativeAmount), + txHash: `0x${'aa'.repeat(32)}` as Hex, units: 1, }), ), diff --git a/src/tempo/session/client/SessionManager.ts b/src/tempo/session/client/SessionManager.ts index 1e2a76f3..f914b96d 100644 --- a/src/tempo/session/client/SessionManager.ts +++ b/src/tempo/session/client/SessionManager.ts @@ -11,12 +11,14 @@ import type { ChannelEntry } from '../client/ChannelOps.js' import { createChannelStore, entryKey, type ChannelStore } from '../client/ChannelStore.js' import type { SessionContext } from '../client/CredentialState.js' import { session as sessionPlugin } from '../client/Session.js' +import * as Channel from '../precompile/Channel.js' import { deserializeSessionReceipt } from '../precompile/Protocol.js' import { readSessionChallengeAmount, type SessionReceipt } from '../precompile/Protocol.js' import { deserializeSnapshot as deserializeSessionSnapshot, serializeSnapshot as serializeSessionSnapshot, } from '../Snapshot.js' +import { registerSessionManagerInternals } from './internal/SessionManager.js' import { createSessionReceiptCoordinator } from './ReceiptCoordinator.js' import { resolveCloseTarget, type CloseTarget } from './Runtime.js' import { assertVoucherWithinLocalLimit as assertVoucherWithinLocalAuthorization } from './Runtime.js' @@ -34,6 +36,7 @@ import { import { closeSocketSession } from './Runtime.js' import { closeHttpSession, + getSessionSnapshot, isTempoSessionChallenge, managementInput, postTopUp, @@ -46,7 +49,7 @@ import { type WebSocketConstructor, WebSocketReadyState, } from './Transports.js' -import { openSseSession, type SseDriverOptions } from './Transports.js' +import { consumeSseSessionResponse, openSseSession, type SseDriverOptions } from './Transports.js' import { applyTopUpResult, resolveManualTopUp, type TopUpRequirement } from './Transports.js' import { openWebSocketSession, @@ -428,8 +431,60 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa }) } - function getValidatedFallbackCloseAmount(target: CloseTarget) { - const closeAmount = getFallbackCloseAmount(target.challenge, target.channelId) + function validateCloseSnapshot(channel: ChannelEntry, challenge: TempoSessionChallenge) { + const snapshot = getSessionSnapshot(challenge) + if (!snapshot) return undefined + + const computedSnapshotId = Channel.computeId({ + ...snapshot.descriptor, + chainId: snapshot.chainId, + escrow: snapshot.escrow, + }) + if (computedSnapshotId.toLowerCase() !== snapshot.channelId.toLowerCase()) { + throw new Error('close snapshot descriptor does not match its channel ID') + } + if (snapshot.channelId.toLowerCase() !== channel.channelId.toLowerCase()) { + throw new Error('close snapshot channel ID does not match local session') + } + if ( + snapshot.chainId !== channel.chainId || + snapshot.escrow.toLowerCase() !== channel.escrow.toLowerCase() + ) { + throw new Error('close snapshot payment scope does not match local session') + } + + const acceptedCumulative = BigInt(snapshot.acceptedCumulative) + const snapshotSpent = BigInt(snapshot.spent) + if (acceptedCumulative < 0n || snapshotSpent < 0n) { + throw new Error('close snapshot amounts must not be negative') + } + if (snapshotSpent > acceptedCumulative) { + throw new Error('close snapshot spent exceeds accepted cumulative amount') + } + if (acceptedCumulative > channel.cumulativeAmount) { + throw new Error('close snapshot accepted cumulative exceeds local voucher state') + } + return { acceptedCumulative, spent: snapshotSpent } + } + + function applyCloseSnapshot(target: CloseTarget, challenge: TempoSessionChallenge) { + const snapshot = validateCloseSnapshot(target.channel, challenge) + if (!snapshot) return + + const { acceptedCumulative, spent } = snapshot + if (runtime.spent > acceptedCumulative) { + throw new Error('close snapshot accepted cumulative is below locally confirmed spend') + } + if (spent > runtime.spent) runtime.spent = spent + } + + function getValidatedFallbackCloseAmount( + target: CloseTarget, + challenge: TempoSessionChallenge = target.challenge, + applySnapshot = false, + ) { + if (applySnapshot) applyCloseSnapshot(target, challenge) + const closeAmount = getFallbackCloseAmount(challenge, target.channelId) if (closeAmount > target.channel.cumulativeAmount) { throw new Error('fallback close amount exceeds local voucher state') } @@ -467,6 +522,7 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa spent: runtime.spent, }) if (applied?.channel && runtime.lastChallenge) { + await store.set(applied.channel) dispatch({ type: 'activated', challengeId: runtime.lastChallenge.id, @@ -623,6 +679,8 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa createSessionCredential, fetch: config.fetch, lastUrl: runtime.lastUrl, + resolveSignedCloseAmount: (challenge) => + getValidatedFallbackCloseAmount(target, challenge, true), signedCloseAmount: getValidatedFallbackCloseAmount(target), setChallenge(challenge) { runtime.lastChallenge = challenge @@ -638,6 +696,20 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa return receipt } + const sseDriver = { + createSessionCredential, + doFetch, + fetch: config.fetch, + getChannel: () => runtime.channel, + getChallenge: () => runtime.lastChallenge, + assertVoucherWithinLocalLimit, + managementInput, + acceptReceipt(receipt: SessionReceipt) { + updateSpentFromReceipt(receipt) + }, + topUpIfNeeded, + } + const self: SessionManager = { get channelId() { return runtime.channel?.channelId @@ -673,19 +745,7 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa }, async sse(input, init) { - return openSseSession(input, init, { - createSessionCredential, - doFetch, - fetch: config.fetch, - getChannel: () => runtime.channel, - getChallenge: () => runtime.lastChallenge, - assertVoucherWithinLocalLimit, - managementInput, - acceptReceipt(receipt) { - updateSpentFromReceipt(receipt) - }, - topUpIfNeeded, - }) + return openSseSession(input, init, sseDriver) }, async ws(input, init) { @@ -766,6 +826,38 @@ export function sessionManager(parameters: sessionManager.Parameters): SessionMa }, } + registerSessionManagerInternals(self, { + consumeSseResponse(input, response, options) { + return consumeSseSessionResponse(input, response, options, sseDriver) + }, + rehydrate({ channel, challenge, input, spent }) { + if (!channel.opened) throw new Error('Cannot restore a closed session channel.') + if (!isTempoSessionChallenge(challenge)) { + throw new Error('Cannot restore session: challenge is not tempo/session.') + } + if (spent < 0n) throw new Error('Cannot restore session: spent must not be negative.') + if (spent > channel.cumulativeAmount) { + throw new Error('Cannot restore session: spent exceeds local voucher state.') + } + assertVoucherWithinLocalLimit(channel.cumulativeAmount) + const snapshot = validateCloseSnapshot(channel, challenge) + const restoredSpent = snapshot && snapshot.spent > spent ? snapshot.spent : spent + + runtime.lastUrl = input + runtime.lastChallenge = challenge + runtime.channel = channel + runtime.spent = restoredSpent + dispatch({ type: 'challengeReceived', challengeId: challenge.id }) + dispatch({ + type: 'activated', + challengeId: challenge.id, + entry: channel, + spent: restoredSpent.toString(), + units: 0, + }) + }, + }) + return self } diff --git a/src/tempo/session/client/Transports.test.ts b/src/tempo/session/client/Transports.test.ts index 75622645..6d6b24f6 100644 --- a/src/tempo/session/client/Transports.test.ts +++ b/src/tempo/session/client/Transports.test.ts @@ -9,6 +9,7 @@ import type { ChannelEntry } from '../client/ChannelOps.js' import type { SessionContext } from '../client/CredentialState.js' import { createSessionReceipt, + formatNeedVoucherEvent, serializeSessionReceipt, tip20ChannelEscrow, type ChannelDescriptor, @@ -18,6 +19,7 @@ import { initialState, type SessionState } from './Runtime.js' import { applyTopUpResult, closeHttpSession, + consumeSseSessionResponse, createActiveSocketSession, driveSseResponse, isExpectedSocketReceipt, @@ -111,9 +113,15 @@ describe('HttpManagement', () => { }) } - function receiptHeader(acceptedCumulative: bigint, spent: bigint) { + function receiptHeader(acceptedCumulative: bigint, spent: bigint, challengeId = 'challenge-1') { return serializeSessionReceipt( - createSessionReceipt({ acceptedCumulative, challengeId: 'challenge-1', channelId, spent }), + createSessionReceipt({ + acceptedCumulative, + challengeId, + channelId, + spent, + txHash: `0x${'aa'.repeat(32)}`, + }), ) } @@ -138,9 +146,12 @@ describe('HttpManagement', () => { ) }) - test('strips resource query state from management URLs', () => { + test('preserves query state and strips fragments from management URLs', () => { expect(managementInput('https://example.test/resource?cursor=abc').toString()).toBe( - 'https://example.test/resource', + 'https://example.test/resource?cursor=abc', + ) + expect(managementInput('https://example.test/resource?cursor=abc#result').toString()).toBe( + 'https://example.test/resource?cursor=abc', ) }) @@ -184,7 +195,7 @@ describe('HttpManagement', () => { return 'top-up-credential' }) const fetch = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => { - expect(input.toString()).toBe('https://example.test/resource') + expect(input.toString()).toBe('https://example.test/resource?cursor=abc') expect(init?.method).toBe('POST') expect(authorizationHeader(init)).toBe('top-up-credential') return new Response(null, { @@ -300,16 +311,20 @@ describe('HttpManagement', () => { const entry = channel() const retryChallenge = { ...challenge(), id: 'challenge-2' } as TempoSessionChallenge const setChallenge = vi.fn() - const createSessionCredential = vi.fn(async (challenge_) => { + const closeAmounts: string[] = [] + const createSessionCredential = vi.fn(async (challenge_, context: SessionContext) => { + if (context.action !== 'close') throw new Error('expected close context') + closeAmounts.push(context.cumulativeAmountRaw!) return `close-${challenge_.id}` }) + const resolveSignedCloseAmount = vi.fn(() => '6') const fetch = vi .fn() .mockResolvedValueOnce(response402(retryChallenge)) .mockResolvedValueOnce( new Response(null, { status: 200, - headers: { [Constants.Headers.paymentReceipt]: receiptHeader(5n, 5n) }, + headers: { [Constants.Headers.paymentReceipt]: receiptHeader(6n, 6n, 'challenge-2') }, }), ) @@ -317,18 +332,71 @@ describe('HttpManagement', () => { createSessionCredential, fetch, lastUrl: 'https://example.test/resource', + resolveSignedCloseAmount, signedCloseAmount: '5', setChallenge, target: { challenge: challenge(), channel: entry, channelId }, }) - expect(receipt?.spent).toBe('5') + expect(receipt?.spent).toBe('6') expect(setChallenge).toHaveBeenCalledWith(retryChallenge) + expect(resolveSignedCloseAmount).toHaveBeenCalledOnce() + expect(resolveSignedCloseAmount).toHaveBeenCalledWith(retryChallenge) + expect(closeAmounts).toEqual(['5', '6']) expect(createSessionCredential).toHaveBeenCalledTimes(2) expect(createSessionCredential.mock.calls[1]?.[0]).toMatchObject({ id: retryChallenge.id }) expect(authorizationHeader(fetch.mock.calls[1]?.[1])).toBe('close-challenge-2') }) + test('closeHttpSession rejects a receipt without settlement proof', async () => { + const receipt = serializeSessionReceipt( + createSessionReceipt({ + acceptedCumulative: 5n, + challengeId: 'challenge-1', + channelId, + spent: 5n, + }), + ) + await expect( + closeHttpSession({ + createSessionCredential: async () => 'close-credential', + fetch: async () => + new Response(null, { + headers: { [Constants.Headers.paymentReceipt]: receipt }, + }), + lastUrl: 'https://example.test/resource', + signedCloseAmount: '5', + target: { challenge: challenge(), channel: channel(), channelId }, + }), + ).rejects.toThrow('Session close response included a mismatched payment receipt.') + }) + + test('closeHttpSession rejects an invalid settlement transaction hash', async () => { + for (const txHash of ['0x1234', `0x${'gg'.repeat(32)}`]) { + const receipt = serializeSessionReceipt( + createSessionReceipt({ + acceptedCumulative: 5n, + challengeId: 'challenge-1', + channelId, + spent: 5n, + txHash: txHash as Hex.Hex, + }), + ) + await expect( + closeHttpSession({ + createSessionCredential: async () => 'close-credential', + fetch: async () => + new Response(null, { + headers: { [Constants.Headers.paymentReceipt]: receipt }, + }), + lastUrl: 'https://example.test/resource', + signedCloseAmount: '5', + target: { challenge: challenge(), channel: channel(), channelId }, + }), + ).rejects.toThrow('Session close response included a mismatched payment receipt.') + } + }) + test('closeHttpSession includes problem detail and challenge header on failure', async () => { await expect( closeHttpSession({ @@ -672,6 +740,80 @@ describe('SseDriver', () => { ]) expect(acceptReceipt).toHaveBeenCalledWith(receipt) }) + + test('keeps voucher events bound to the challenge that opened the stream', async () => { + const channelId = `0x${'01'.repeat(32)}` as Hex.Hex + const token = '0x20c0000000000000000000000000000000000001' as Address + const payee = '0x0000000000000000000000000000000000000002' as Address + const descriptor: ChannelDescriptor = { + authorizedSigner: '0x0000000000000000000000000000000000000001', + expiringNonceHash: `0x${'03'.repeat(32)}` as Hex.Hex, + operator: '0x0000000000000000000000000000000000000000', + payee, + payer: '0x0000000000000000000000000000000000000001', + salt: `0x${'02'.repeat(32)}` as Hex.Hex, + token, + } + const makeChallenge = (id: string) => + Challenge.from({ + id, + intent: Constants.Intents.session, + method: Constants.Methods.tempo, + realm: 'example.test', + request: { + amount: '1', + currency: token, + methodDetails: { chainId: 4217, escrowContract: tip20ChannelEscrow }, + recipient: payee, + unitType: 'request', + }, + }) as TempoSessionChallenge + const streamChallenge = makeChallenge('challenge-1') + let currentChallenge = streamChallenge + const createSessionCredential = vi.fn(async () => 'Payment credential') + const channel: ChannelEntry = { + chainId: 4217, + channelId, + cumulativeAmount: 1n, + deposit: 10n, + descriptor, + escrow: tip20ChannelEscrow, + opened: true, + } + const response = new Response( + formatNeedVoucherEvent({ + acceptedCumulative: '1', + channelId, + deposit: '10', + requiredCumulative: '2', + }), + { headers: { 'Content-Type': 'text/event-stream' } }, + ) + const stream = consumeSseSessionResponse('https://example.test/stream', response, undefined, { + acceptReceipt() {}, + assertVoucherWithinLocalLimit() {}, + createSessionCredential, + async doFetch() { + throw new Error('unexpected resource fetch') + }, + fetch: vi.fn(async () => new Response(null, { status: 204 })), + getChallenge: () => currentChallenge, + getChannel: () => channel, + managementInput: (input) => input, + async topUpIfNeeded() {}, + }) + + currentChallenge = makeChallenge('challenge-2') + const frames: string[] = [] + for await (const frame of stream) frames.push(frame) + + expect(frames).toEqual([]) + expect(createSessionCredential).toHaveBeenCalledOnce() + expect(createSessionCredential).toHaveBeenCalledWith( + streamChallenge, + expect.objectContaining({ action: 'voucher', cumulativeAmountRaw: '2' }), + ) + }) }) describe('WsDriver', () => { diff --git a/src/tempo/session/client/Transports.ts b/src/tempo/session/client/Transports.ts index 47b52481..f26bacca 100644 --- a/src/tempo/session/client/Transports.ts +++ b/src/tempo/session/client/Transports.ts @@ -1,4 +1,4 @@ -import type { Hex } from 'ox' +import { Hex } from 'ox' import * as Challenge from '../../../Challenge.js' import * as Fetch from '../../../client/internal/Fetch.js' @@ -502,6 +502,8 @@ export type CloseHttpSessionParameters = { fetch: typeof globalThis.fetch /** Last paid resource URL; used as the management endpoint base. */ lastUrl: RequestInfo | URL | null + /** Resolves the close amount again when the server refreshes the challenge. */ + resolveSignedCloseAmount?: ((challenge: TempoSessionChallenge) => string) | undefined /** Final cumulative amount the client is willing to sign. */ signedCloseAmount: string /** Stores a fresh close challenge when the first close credential expired. */ @@ -544,7 +546,7 @@ export function requestInitWithAuthorization( } } -/** Returns the URL used for out-of-band management POSTs, stripping resource query state. */ +/** Returns the exact resource URL used for out-of-band management POSTs, without its fragment. */ export function managementInput(input: RequestInfo | URL): RequestInfo | URL { try { const base = @@ -555,7 +557,7 @@ export function managementInput(input: RequestInfo | URL): RequestInfo | URL { : input instanceof URL ? new URL(input) : new URL(String(input), base) - url.search = '' + url.hash = '' return url } catch { return input @@ -682,12 +684,18 @@ export async function closeHttpSession( ) } - const postClose = async (challenge: TempoSessionChallenge) => { + let closeChallenge = parameters.target.challenge + let signedCloseAmount = parameters.signedCloseAmount + const postClose = async (challenge: TempoSessionChallenge, refreshed = false) => { + closeChallenge = challenge + signedCloseAmount = + (refreshed ? parameters.resolveSignedCloseAmount?.(challenge) : undefined) ?? + parameters.signedCloseAmount const credential = await parameters.createSessionCredential(challenge, { action: 'close', channelId: parameters.target.channelId, descriptor: parameters.target.channel.descriptor, - cumulativeAmountRaw: parameters.signedCloseAmount, + cumulativeAmountRaw: signedCloseAmount, }) return parameters.fetch(parameters.lastUrl!, { method: 'POST', @@ -700,7 +708,7 @@ export async function closeHttpSession( const challenge = Challenge.fromResponseList(response).find(isTempoSessionChallenge) if (challenge) { parameters.setChallenge?.(challenge) - response = await postClose(challenge) + response = await postClose(challenge, true) } } if (!response.ok) { @@ -712,7 +720,21 @@ export async function closeHttpSession( } const receiptHeader = response.headers.get(Constants.Headers.paymentReceipt) - return receiptHeader ? deserializeSessionReceipt(receiptHeader) : undefined + if (!receiptHeader) return undefined + const receipt = deserializeSessionReceipt(receiptHeader) + if ( + !receipt.txHash || + !Hex.validate(receipt.txHash, { strict: true }) || + Hex.size(receipt.txHash) !== 32 || + !isExpectedCloseReceipt({ + challengeId: closeChallenge.id, + channelId: parameters.target.channelId, + expectedCloseAmount: signedCloseAmount, + receipt, + }) + ) + throw new Error('Session close response included a mismatched payment receipt.') + return receipt } /** Options accepted by the auto-driving SSE session flow. */ @@ -723,6 +745,12 @@ export type SseDriverOptions = RequestInit & { signal?: AbortSignal | undefined } +/** @internal Options used when consuming an already-paid SSE response. */ +export type SseResponseOptions = { + onReceipt?: ((receipt: SessionReceipt) => void) | undefined + signal?: AbortSignal | undefined +} + /** Dependencies the SSE driver needs from `SessionManager`. */ export type OpenSseSessionParameters = { /** Creates a session credential for the selected challenge/context. */ @@ -773,18 +801,28 @@ export async function openSseSession( if (!isEventStream(response) && driver.getChannel()?.opened) response = await driver.doFetch(input, sseInit) - const challenge = driver.getChallenge() + return consumeSseSessionResponse(input, response, { onReceipt, signal }, driver) +} + +/** @internal Consumes an already-paid SSE response with the session transport driver. */ +export function consumeSseSessionResponse( + input: RequestInfo | URL, + response: Response, + options: SseResponseOptions | undefined, + driver: OpenSseSessionParameters, +): AsyncIterable { if (!isEventStream(response)) throw new Error('SSE response is not an event stream.') if (!response.body) throw new Error('Response has no body.') + const challenge = driver.getChallenge() return iterateSseMessages({ onNeedVoucher: (event) => handleSseNeedVoucher({ challenge, driver, input }, event), onReceipt(receipt) { driver.acceptReceipt(receipt) - onReceipt?.(receipt) + options?.onReceipt?.(receipt) }, response, - signal, + signal: options?.signal, }) } diff --git a/src/tempo/session/client/internal/SessionManager.ts b/src/tempo/session/client/internal/SessionManager.ts new file mode 100644 index 00000000..914ef023 --- /dev/null +++ b/src/tempo/session/client/internal/SessionManager.ts @@ -0,0 +1,36 @@ +import type { ChannelEntry } from '../ChannelOps.js' +import type { PaymentResponse, SessionManager } from '../SessionManager.js' +import type { SseResponseOptions, TempoSessionChallenge } from '../Transports.js' + +type RehydrateParameters = { + channel: ChannelEntry + challenge: TempoSessionChallenge + input: RequestInfo | URL + spent: bigint +} + +type SessionManagerInternals = { + consumeSseResponse( + input: RequestInfo | URL, + response: PaymentResponse, + options?: SseResponseOptions | undefined, + ): AsyncIterable + rehydrate(parameters: RehydrateParameters): void +} + +const internals = new WeakMap() + +/** @internal Registers private transport and recovery hooks for a session manager. */ +export function registerSessionManagerInternals( + manager: SessionManager, + value: SessionManagerInternals, +): void { + internals.set(manager, value) +} + +/** @internal Returns private transport and recovery hooks for a session manager. */ +export function getSessionManagerInternals(manager: SessionManager): SessionManagerInternals { + const value = internals.get(manager) + if (!value) throw new Error('Session manager internals are unavailable.') + return value +}