diff --git a/apps/server/src/loopback-server.ts b/apps/server/src/loopback-server.ts index 6dc00a9..8b2989e 100644 --- a/apps/server/src/loopback-server.ts +++ b/apps/server/src/loopback-server.ts @@ -10,44 +10,28 @@ import { PROTOCOL_VERSION, PROTOCOL_VERSION_HEADER, PROTOCOL_LIMITS, - redactDiagnostic, + createDiagnosticRedactor, sseInvalidationSchema, type HealthResponse, + type DiagnosticRedactor, type ProtocolError, - type SessionMetadata, } from '@codex-git/protocol'; +import { + createProtocolDispatcher, + PROTOCOL_ENDPOINTS, + redactProtocolDiagnostics, + resolveProtocolEndpoint, + type ProtocolDispatcher, + type ProtocolHandlers, +} from './protocol-dispatch.js'; + const healthResponse = { product: 'codex-git', status: 'ok', } satisfies HealthResponse; const loopbackHost = '127.0.0.1' as const; -const endpointMethods = new Map([ - ['branches', 'POST'], - ['commands', 'POST'], - ['diff', 'POST'], - ['draft', 'PUT'], - ['events', 'GET'], - ['native-actions', 'POST'], - ['operations', 'POST'], - ['session', 'GET'], - ['snapshot', 'GET'], -]); -const sessionMetadata = { - protocolVersion: PROTOCOL_VERSION, - capabilities: { - branchSearch: false, - commands: false, - commitDrafts: false, - diff: false, - events: true, - nativeActions: false, - operationRecovery: false, - }, - limits: PROTOCOL_LIMITS, -} satisfies SessionMetadata; - export interface LoopbackAddress { readonly host: typeof loopbackHost; readonly port: number; @@ -65,6 +49,7 @@ export interface LoopbackServer { export interface LoopbackServerOptions { readonly allowedOrigins?: readonly string[]; + readonly handlers?: ProtocolHandlers; readonly randomBytes?: (length: number) => Uint8Array; } @@ -78,6 +63,8 @@ export async function startLoopbackServer( const token = Buffer.from(tokenBytes).toString('hex'); const instancePrefix = `/instance/${token}/v1`; const allowedOrigins = new Set(options.allowedOrigins ?? []); + const redactDiagnostic = createDiagnosticRedactor({ secrets: [token] }); + const dispatcher = createProtocolDispatcher(options.handlers); const events = createEventBroker(); const server = createServer((request, response) => { void handleRequest( @@ -86,6 +73,8 @@ export async function startLoopbackServer( token, instancePrefix, allowedOrigins, + redactDiagnostic, + dispatcher, events, ).catch(() => { if (!response.headersSent) { @@ -138,6 +127,8 @@ async function handleRequest( token: string, instancePrefix: string, allowedOrigins: ReadonlySet, + redactDiagnostic: DiagnosticRedactor, + dispatcher: ProtocolDispatcher, events: EventBroker, ): Promise { if (request.socket.remoteAddress !== loopbackHost) { @@ -183,16 +174,18 @@ async function handleRequest( return; } - const endpoint = pathname.startsWith(`${instancePrefix}/`) + const endpointName = pathname.startsWith(`${instancePrefix}/`) ? pathname.slice(instancePrefix.length + 1) : ''; - const expectedMethod = endpointMethods.get(endpoint); + const endpoint = resolveProtocolEndpoint(endpointName); + const expectedMethod = + endpoint === undefined ? undefined : PROTOCOL_ENDPOINTS[endpoint].method; if (request.method === 'OPTIONS') { handlePreflight(request, response, origin, expectedMethod, allowedOrigins); return; } - if (endpoint !== 'events' && version !== String(PROTOCOL_VERSION)) { + if (endpointName !== 'events' && version !== String(PROTOCOL_VERSION)) { sendProtocolError(response, 426, { code: 'unsupported_protocol_version', details: { @@ -212,7 +205,7 @@ async function handleRequest( return; } - if (expectedMethod === undefined) { + if (endpoint === undefined) { sendProtocolError( response, 404, @@ -234,6 +227,7 @@ async function handleRequest( return; } + let body: Buffer | undefined; if (expectedMethod === 'POST' || expectedMethod === 'PUT') { const mediaType = request.headers['content-type']?.split(';', 1)[0]?.trim(); if (mediaType !== 'application/json') { @@ -249,23 +243,26 @@ async function handleRequest( ); return; } - const body = await readBoundedBody( + const requestBody = await readBoundedBody( request, - sessionMetadata.limits.requestBodyBytes, + dispatcher.sessionMetadata.limits.requestBodyBytes, ); - if (body === null) { + if (requestBody === null) { sendProtocolError( response, 413, { code: 'body_too_large', - details: { limitBytes: sessionMetadata.limits.requestBodyBytes }, + details: { + limitBytes: dispatcher.sessionMetadata.limits.requestBodyBytes, + }, message: 'The request body exceeds the configured limit.', }, origin, ); return; } + body = requestBody; } if (endpoint === 'events') { @@ -274,7 +271,32 @@ async function handleRequest( } if (endpoint === 'session') { - sendJson(response, 200, sessionMetadata, origin); + sendJson(response, 200, dispatcher.sessionMetadata, origin); + return; + } + + try { + const dispatched = await dispatcher.dispatch(endpoint, body); + if (dispatched !== undefined) { + sendJson( + response, + dispatched.status, + redactProtocolDiagnostics(dispatched.value, redactDiagnostic), + origin, + PROTOCOL_ENDPOINTS[endpoint].responseBodyBytes, + ); + return; + } + } catch { + sendProtocolError( + response, + 500, + { + code: 'internal_error', + message: 'The protocol request could not be completed.', + }, + origin, + ); return; } @@ -503,9 +525,10 @@ function sendJson( status: number, value: unknown, origin?: string, + limitBytes: number = PROTOCOL_LIMITS.diffOutputBytes, ): void { const body = JSON.stringify(value); - if (Buffer.byteLength(body) > sessionMetadata.limits.diffOutputBytes) { + if (Buffer.byteLength(body) > limitBytes) { response.writeHead(507, { ...(origin === undefined ? {} @@ -516,7 +539,7 @@ function sendJson( JSON.stringify({ error: { code: 'output_too_large', - details: { limitBytes: sessionMetadata.limits.diffOutputBytes }, + details: { limitBytes }, message: 'The protocol response exceeds the configured limit.', }, }), diff --git a/apps/server/src/protocol-dispatch.test.ts b/apps/server/src/protocol-dispatch.test.ts new file mode 100644 index 0000000..bd346d5 --- /dev/null +++ b/apps/server/src/protocol-dispatch.test.ts @@ -0,0 +1,797 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +import { + branchSearchResultSchema, + commitDraftSchema, + PROTOCOL_LIMITS, + PROTOCOL_VERSION_HEADER, + diffResultSchema, + operationReceiptSchema, + operationResultSchema, + repositorySnapshotSchema, +} from '@codex-git/protocol'; + +import { startLoopbackServer, type LoopbackServer } from './server.js'; + +const servers: LoopbackServer[] = []; +const headers = { + origin: 'null', + [PROTOCOL_VERSION_HEADER]: '1', +} as const; +const staleTargetResponse = { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, +}; + +afterEach(async () => { + await Promise.all(servers.splice(0).map((server) => server.close())); +}); + +describe('protocol HTTP dispatch', () => { + it('routes snapshots with redacted diagnostics without advertising absent handlers', async () => { + const snapshot = repositorySnapshotSchema.parse({ + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + repositoryRevision: 4, + topologyRevision: 2, + refsRevision: 3, + refresh: { + kind: 'stale', + message: 'Authorization: Bearer fixture-snapshot-token', + }, + worktrees: [ + { + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + worktreeRevision: 1, + generation: 'generation_0123456789abcdef0123456789abcdef', + freshness: { + kind: 'failed', + message: + 'remote=https://alice:fixture-password@example.com/repo.git', + }, + head: { kind: 'initial' }, + indexTree: null, + status: { + kind: 'unavailable', + reason: 'token=fixture-unavailable-secret', + }, + changes: Array.from({ length: 2_000 }, (_, index) => ({ + baseline: 'index_to_working_tree', + displayPath: 'x'.repeat(4_096), + fileId: `file_${index.toString(16).padStart(32, '0')}`, + kind: 'change', + nativeTargets: [], + })), + nativeTargets: [], + }, + ], + remotes: ['a.co', 'a.co:1', '1.1.1.1', '[::1]'].map((host, index) => ({ + remoteId: `remote_${index.toString(16).padStart(32, '0')}`, + displayName: host, + host, + })), + operations: [], + }); + const handleSnapshot = vi.fn(() => snapshot); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { snapshot: handleSnapshot }, + }); + servers.push(server); + + const [sessionResponse, snapshotResponse] = await Promise.all([ + fetch(server.sessionUrl, { headers }), + fetch(endpointUrl(server, 'snapshot'), { headers }), + ]); + + const snapshotBody = await snapshotResponse.json(); + expect(await sessionResponse.json()).toMatchObject({ + capabilities: { + branchSearch: false, + commands: false, + commitDrafts: false, + diff: false, + events: true, + nativeActions: false, + operationRecovery: false, + }, + }); + expect(repositorySnapshotSchema.safeParse(snapshotBody).success).toBe(true); + expect(snapshotResponse.status).toBe(200); + expect(snapshotBody).toMatchObject({ + refresh: { message: 'Authorization: [REDACTED]' }, + worktrees: [ + { + freshness: { + message: 'remote=https://[REDACTED]@example.com/repo.git', + }, + status: { reason: 'token=[REDACTED]' }, + }, + ], + }); + expect(JSON.stringify(snapshotBody)).not.toMatch( + /fixture-(?:snapshot-token|password|unavailable-secret)/u, + ); + + handleSnapshot.mockReturnValueOnce({ + ...snapshot, + remotes: [ + { ...snapshot.remotes[0]!, host: 'https://a:fixture@example.com' }, + ], + }); + const invalidHostResponse = await fetch(endpointUrl(server, 'snapshot'), { + headers, + }); + const invalidHostBody = + await expectInvalidHandlerResponse(invalidHostResponse); + expect(JSON.stringify(invalidHostBody)).not.toContain('a:fixture'); + expect(handleSnapshot).toHaveBeenCalledTimes(2); + }); + + it('rejects malformed diff requests before calling the installed handler', async () => { + const handleDiff = vi.fn(() => + diffResultSchema.parse({ + kind: 'binary' as const, + fileId: 'file_0123456789abcdef0123456789abcdef', + baseline: 'index_to_working_tree' as const, + byteCount: 42, + }), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { diff: handleDiff }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'diff'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + fileId: 'file_0123456789abcdef0123456789abcdef', + path: '/Users/example/repository/private.txt', + }), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { diff: boolean } }; + + expect({ + capability: session.capabilities.diff, + response: await response.json(), + status: response.status, + }).toEqual({ + capability: true, + response: { + error: { + code: 'invalid_payload', + details: expect.objectContaining({ issues: expect.any(Array) }), + message: 'The protocol payload is invalid.', + }, + }, + status: 400, + }); + expect(handleDiff).not.toHaveBeenCalled(); + }); + + it('rejects malformed UTF-8 JSON bytes before calling a handler', async () => { + const searchBranches = vi.fn(() => + branchSearchResultSchema.parse({ refsRevision: 1, candidates: [] }), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { branchSearch: searchBranches }, + }); + servers.push(server); + const prefix = new TextEncoder().encode( + `{"worktreeId":"worktree_0123456789abcdef0123456789abcdef","query":"`, + ); + const suffix = new TextEncoder().encode('"}'); + const body = Uint8Array.from([...prefix, 0xc3, 0x28, ...suffix]); + + const response = await fetch(endpointUrl(server, 'branches'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body, + }); + + expect({ body: await response.json(), status: response.status }).toEqual({ + body: { + error: { + code: 'invalid_payload', + message: 'The protocol request body is not valid JSON.', + }, + }, + status: 400, + }); + expect(searchBranches).not.toHaveBeenCalled(); + }); + + it('does not serialize a handler response that fails runtime validation', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: async () => + ({ + kind: 'text', + fileId: 'file_0123456789abcdef0123456789abcdef', + baseline: 'head_to_index', + content: 'token=fixture-handler-secret', + lineCount: 20_001, + }) as never, + }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'diff'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + fileId: 'file_0123456789abcdef0123456789abcdef', + }), + }); + const body = JSON.stringify(await expectInvalidHandlerResponse(response)); + expect(body).not.toContain('fixture-handler-secret'); + }); + + it('rejects a valid diff response for a different opaque File ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: async () => + diffResultSchema.parse({ + kind: 'binary', + fileId: 'file_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + baseline: 'head_to_index', + byteCount: 24, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'diff', { + fileId: 'file_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + }); + + await expectInvalidHandlerResponse(response); + }); + + it('returns a diff at the exact negotiated content limit', async () => { + const fileId = 'file_cccccccccccccccccccccccccccccccc'; + const content = 'x'.repeat(PROTOCOL_LIMITS.diffOutputBytes); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + diff: () => + diffResultSchema.parse({ + kind: 'text', + fileId, + baseline: 'head_to_index', + content, + lineCount: 1, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'diff', { fileId }); + const body = (await response.json()) as { content?: string }; + + expect({ + contentBytes: body.content?.length, + status: response.status, + }).toEqual({ + contentBytes: PROTOCOL_LIMITS.diffOutputBytes, + status: 200, + }); + }); + + it('dispatches branch search with validated opaque targets', async () => { + const result = branchSearchResultSchema.parse({ + refsRevision: 7, + candidates: Array.from({ length: 5_000 }, (_, index) => ({ + refId: `ref_${index.toString(16).padStart(32, '0')}`, + kind: 'local' as const, + displayName: 'x'.repeat(1_024), + occupiedBy: null, + })), + }); + const searchBranches = vi.fn(() => result); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { branchSearch: searchBranches }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'branches'), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + query: 'feature', + }), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { branchSearch: boolean } }; + + expect(session.capabilities.branchSearch).toBe(true); + expect(await response.json()).toEqual(result); + expect(response.status).toBe(200); + expect(searchBranches).toHaveBeenCalledWith({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + query: 'feature', + }); + }); + + it('dispatches a validated Commit Draft update through the installed handler', async () => { + const draft = commitDraftSchema.parse({ + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + revision: 9, + text: 'Describe the protocol change', + }); + const updateDraft = vi.fn(() => draft); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commitDrafts: updateDraft }, + }); + servers.push(server); + const request = { + worktreeId: 'worktree_0123456789abcdef0123456789abcdef', + expectedRevision: 8, + update: { kind: 'set', text: 'Describe the protocol change' }, + } as const; + + const response = await fetch(endpointUrl(server, 'draft'), { + method: 'PUT', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify(request), + }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { commitDrafts: boolean } }; + + expect({ + capability: session.capabilities.commitDrafts, + result: await response.json(), + status: response.status, + }).toEqual({ capability: true, result: draft, status: 200 }); + expect(updateDraft).toHaveBeenCalledWith(request); + }); + + it('rejects a Commit Draft response for a different opaque Worktree ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + commitDrafts: async () => + commitDraftSchema.parse({ + worktreeId: 'worktree_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb', + revision: 3, + text: 'Wrong Worktree draft', + }), + }, + }); + servers.push(server); + + const response = await fetch(endpointUrl(server, 'draft'), { + method: 'PUT', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify({ + worktreeId: 'worktree_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', + expectedRevision: 2, + update: { kind: 'set', text: 'Expected Worktree draft' }, + }), + }); + + await expectInvalidHandlerResponse(response); + }); + + it('dispatches one validated Product Command and returns its receipt', async () => { + const request = { + clientCommandId: 'command_0123456789abcdef0123456789abcdef', + command: { + kind: 'refresh', + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + }, + } as const; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_0123456789abcdef0123456789abcdef', + clientCommandId: request.clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(() => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const response = await postJson(server, 'commands', request); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { commands: boolean } }; + + expect({ + capability: session.capabilities.commands, + receipt: await response.json(), + status: response.status, + }).toEqual({ capability: true, receipt, status: 200 }); + expect(dispatchCommand).toHaveBeenCalledWith(request); + }); + + it('returns one operation for concurrent exact command retries', async () => { + const request = { + clientCommandId: 'command_11111111111111111111111111111111', + command: { + kind: 'refresh', + repositoryId: 'repository_0123456789abcdef0123456789abcdef', + }, + } as const; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_11111111111111111111111111111111', + clientCommandId: request.clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(async () => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const responses = await Promise.all([ + postJson(server, 'commands', request), + postJson(server, 'commands', { + clientCommandId: request.clientCommandId, + command: { + repositoryId: request.command.repositoryId, + kind: request.command.kind, + }, + }), + ]); + const bodies = await Promise.all( + responses.map((response) => response.json()), + ); + + expect(bodies).toEqual([receipt, { ...receipt, disposition: 'duplicate' }]); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('rejects a command ID reused for a different validated intent', async () => { + const clientCommandId = 'command_22222222222222222222222222222222'; + const receipt = operationReceiptSchema.parse({ + operationId: 'operation_22222222222222222222222222222222', + clientCommandId, + disposition: 'accepted', + }); + const dispatchCommand = vi.fn(() => receipt); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + await postJson(server, 'commands', { + clientCommandId, + command: { + kind: 'refresh', + repositoryId: 'repository_11111111111111111111111111111111', + }, + }); + const collision = await postJson(server, 'commands', { + clientCommandId, + command: { + kind: 'refresh', + repositoryId: 'repository_22222222222222222222222222222222', + }, + }); + + expect({ body: await collision.json(), status: collision.status }).toEqual({ + body: { + error: { + code: 'command_id_collision', + message: + 'The client command ID was already used for another command.', + }, + }, + status: 409, + }); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('records a command before a synchronously failing handler can be retried', async () => { + const request = { + clientCommandId: 'command_77777777777777777777777777777777', + command: { + kind: 'refresh', + repositoryId: 'repository_77777777777777777777777777777777', + }, + } as const; + const dispatchCommand = vi.fn(() => { + throw new Error('fixture handler failure'); + }); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { commands: dispatchCommand }, + }); + servers.push(server); + + const first = await postJson(server, 'commands', request); + const retry = await postJson(server, 'commands', request); + + expect({ + allowOrigin: first.headers.get('access-control-allow-origin'), + first: await first.json(), + firstStatus: first.status, + retry: await retry.json(), + retryStatus: retry.status, + }).toEqual({ + allowOrigin: 'null', + first: { + error: { + code: 'internal_error', + message: 'The protocol request could not be completed.', + }, + }, + firstStatus: 500, + retry: { + error: { + code: 'duplicate_command', + message: 'The duplicate command cannot be replayed safely.', + }, + }, + retryStatus: 409, + }); + expect(dispatchCommand).toHaveBeenCalledOnce(); + }); + + it('routes operation recovery by opaque Operation ID', async () => { + const operationId = 'operation_33333333333333333333333333333333'; + const launchToken = 'ab'.repeat(32); + const diagnostic = 'Authorization: Bearer fixture-operation-token'; + const padding = 'x'.repeat(8_192 - launchToken.length); + const result = operationResultSchema.parse({ + kind: 'partial_success', + operationId, + message: `${launchToken}${padding}`, + effects: [ + { kind: 'succeeded', label: 'x'.repeat(256) }, + ...Array.from({ length: 999 }, (_, index) => ({ + kind: 'failed_known', + label: 'x'.repeat(256), + code: 'authentication', + message: index === 0 ? diagnostic : 'x'.repeat(8_192), + })), + ], + }); + const recoverOperation = vi.fn(() => result); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + randomBytes: (length) => new Uint8Array(length).fill(0xab), + handlers: { operationRecovery: recoverOperation }, + }); + servers.push(server); + + const response = await postJson(server, 'operations', { operationId }); + const session = (await ( + await fetch(server.sessionUrl, { headers }) + ).json()) as { capabilities: { operationRecovery: boolean } }; + + const responseBody = operationResultSchema.parse(await response.json()); + expect(session.capabilities.operationRecovery).toBe(true); + expect(response.status).toBe(200); + if (responseBody.kind !== 'partial_success') throw new Error('unreachable'); + expect(responseBody.message).toBe(`[REDACTED]${padding}`); + expect(responseBody.effects).toHaveLength(1_000); + expect(responseBody.effects[1]).toMatchObject({ + message: 'Authorization: [REDACTED]', + }); + const serialized = JSON.stringify(responseBody); + expect(serialized).not.toContain(launchToken); + expect(serialized).not.toContain('fixture-operation-token'); + expect(recoverOperation).toHaveBeenCalledWith(operationId); + }); + + it('rejects a recovery result for a different Operation ID', async () => { + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + operationRecovery: async () => + operationResultSchema.parse({ + kind: 'succeeded', + operationId: 'operation_55555555555555555555555555555555', + result: { kind: 'no_change' }, + }), + }, + }); + servers.push(server); + + const response = await postJson(server, 'operations', { + operationId: 'operation_44444444444444444444444444444444', + }); + + await expectInvalidHandlerResponse(response); + }); + + it('rejects a fabricated native target that was not issued by the snapshot', async () => { + const issuedTargetId = 'native_66666666666666666666666666666666'; + const perform = vi.fn(() => ({ kind: 'performed' as const })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(issuedTargetId, ['copy_relative_path']), + nativeActions: perform, + }, + }); + servers.push(server); + + await fetch(endpointUrl(server, 'snapshot'), { headers }); + const response = await postJson(server, 'native-actions', { + kind: 'copy_relative_path', + targetId: 'native_99999999999999999999999999999999', + }); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: staleTargetResponse, + status: 409, + }); + expect(perform).not.toHaveBeenCalled(); + }); + + it('rejects an action that was not issued for the native target', async () => { + const targetId = 'native_77777777777777777777777777777777'; + const actions = ['copy_relative_path', 'reveal_in_finder'] as const; + let duplicate: readonly string[] | undefined; + const perform = vi.fn(() => ({ kind: 'performed' as const })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(targetId, actions, duplicate), + nativeActions: perform, + }, + }); + servers.push(server); + + const snapshotUrl = endpointUrl(server, 'snapshot'); + await fetch(snapshotUrl, { headers }); + const response = await postJson(server, 'native-actions', { + kind: 'open_default_app', + targetId, + }); + + expect(await response.json()).toEqual(staleTargetResponse); + expect(response.status).toBe(409); + expect(perform).not.toHaveBeenCalled(); + + duplicate = ['reveal_in_finder', 'copy_relative_path']; + const identicalResponse = await fetch(snapshotUrl, { headers }); + expect(identicalResponse.status).toBe(200); + duplicate = ['copy_relative_path']; + const conflictingResponse = await fetch(snapshotUrl, { headers }); + await expectInvalidHandlerResponse(conflictingResponse); + }); + + it('performs an allow-listed native action against its exact opaque target', async () => { + const targetId = 'native_88888888888888888888888888888888'; + const perform = vi.fn(() => ({ + kind: 'copy_text' as const, + text: 'src/protocol-dispatch.ts', + })); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(targetId, ['copy_relative_path']), + nativeActions: perform, + }, + }); + servers.push(server); + const request = { kind: 'copy_relative_path', targetId } as const; + + await fetch(endpointUrl(server, 'snapshot'), { headers }); + const response = await postJson(server, 'native-actions', request); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: { kind: 'copy_text', text: 'src/protocol-dispatch.ts' }, + status: 200, + }); + expect(perform).toHaveBeenCalledWith(request); + }); + + it('leaves current-state validation and execution in one native handler call', async () => { + const targetId = 'native_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'; + let targetExists = true; + const perform = vi.fn(() => + targetExists + ? ({ kind: 'performed' } as const) + : ({ + kind: 'unavailable', + message: 'The target no longer exists.', + } as const), + ); + const server = await startLoopbackServer({ + allowedOrigins: ['null'], + handlers: { + snapshot: () => nativeSnapshot(targetId, ['reveal_in_finder']), + nativeActions: perform, + }, + }); + servers.push(server); + + await fetch(endpointUrl(server, 'snapshot'), { headers }); + targetExists = false; + const response = await postJson(server, 'native-actions', { + kind: 'reveal_in_finder', + targetId, + }); + + expect({ result: await response.json(), status: response.status }).toEqual({ + result: { kind: 'unavailable', message: 'The target no longer exists.' }, + status: 200, + }); + expect(perform).toHaveBeenCalledOnce(); + }); +}); + +function endpointUrl(server: LoopbackServer, endpoint: string): URL { + const url = new URL(server.sessionUrl); + url.pathname = url.pathname.replace(/\/session$/u, `/${endpoint}`); + return url; +} + +function postJson( + server: LoopbackServer, + endpoint: string, + value: unknown, +): Promise { + return fetch(endpointUrl(server, endpoint), { + method: 'POST', + headers: { ...headers, 'content-type': 'application/json' }, + body: JSON.stringify(value), + }); +} + +async function expectInvalidHandlerResponse(response: Response) { + const body = await response.json(); + expect({ body, status: response.status }).toEqual({ + body: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + status: 500, + }); + return body; +} + +function nativeSnapshot( + targetId: string, + actions: readonly string[], + duplicate?: readonly string[], +): ReturnType { + return repositorySnapshotSchema.parse({ + repositoryId: 'repository_99999999999999999999999999999999', + repositoryRevision: 1, + topologyRevision: 1, + refsRevision: 1, + refresh: { kind: 'current' }, + worktrees: [ + { + worktreeId: 'worktree_99999999999999999999999999999999', + worktreeRevision: 1, + generation: 'generation_99999999999999999999999999999999', + freshness: { kind: 'current' }, + head: { kind: 'initial' }, + indexTree: null, + status: { kind: 'clean' }, + changes: [], + nativeTargets: [ + { targetId, actions }, + ...(duplicate ? [{ targetId, actions: duplicate }] : []), + ], + }, + ], + remotes: [], + operations: [], + }); +} diff --git a/apps/server/src/protocol-dispatch.ts b/apps/server/src/protocol-dispatch.ts new file mode 100644 index 0000000..9326b07 --- /dev/null +++ b/apps/server/src/protocol-dispatch.ts @@ -0,0 +1,465 @@ +import { createHash } from 'node:crypto'; + +import { + branchSearchRequestSchema, + branchSearchResultSchema, + commitDraftSchema, + commitDraftUpdateSchema, + commandEnvelopeSchema, + diffRequestSchema, + diffResultSchema, + operationReceiptSchema, + operationRecoveryRequestSchema, + operationResultSchema, + nativeActionRequestSchema, + nativeActionResultSchema, + parseProtocolPayload, + PROTOCOL_LIMITS, + PROTOCOL_VERSION, + type DiffRequest, + type DiffResult, + type FileId, + type BranchSearchRequest, + type BranchSearchResult, + type CommitDraft, + type CommitDraftUpdate, + type CommandEnvelope, + type RepositorySnapshot, + type OperationReceipt, + type OperationId, + type OperationResult, + type NativeActionKind, + type NativeActionRequest, + type NativeActionResult, + type NativeTargetId, + repositorySnapshotSchema, + type SessionMetadata, + type WorktreeId, + type DiagnosticRedactor, +} from '@codex-git/protocol'; + +type Awaitable = Promise | T; + +export type NativeActionHandler = ( + request: NativeActionRequest, +) => Awaitable; + +export interface ProtocolHandlers { + readonly branchSearch?: ( + request: BranchSearchRequest, + ) => Awaitable; + readonly commitDrafts?: ( + request: CommitDraftUpdate, + ) => Awaitable; + readonly commands?: (request: CommandEnvelope) => Awaitable; + readonly operationRecovery?: ( + operationId: OperationId, + ) => Awaitable; + readonly nativeActions?: NativeActionHandler; + readonly diff?: (request: DiffRequest) => Awaitable; + readonly snapshot?: () => Awaitable; +} + +export interface ProtocolDispatchResponse { + readonly status: number; + readonly value: unknown; +} + +const defaultResponseBodyBytes = PROTOCOL_LIMITS.diffOutputBytes; +const boundedEnvelope = (textBytes: number) => + defaultResponseBodyBytes + textBytes * 6; +const branchResponseBodyBytes = boundedEnvelope(5_000 * 1_024); +const diffResponseBodyBytes = boundedEnvelope(PROTOCOL_LIMITS.diffOutputBytes); +const operationResponseBodyBytes = boundedEnvelope(1_000 * (8_192 + 256)); +const snapshotResponseBodyBytes = boundedEnvelope(2_000 * 4_096); + +export const PROTOCOL_ENDPOINTS = { + branches: { method: 'POST', responseBodyBytes: branchResponseBodyBytes }, + commands: { method: 'POST', responseBodyBytes: defaultResponseBodyBytes }, + diff: { method: 'POST', responseBodyBytes: diffResponseBodyBytes }, + draft: { method: 'PUT', responseBodyBytes: defaultResponseBodyBytes }, + events: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, + 'native-actions': { + method: 'POST', + responseBodyBytes: defaultResponseBodyBytes, + }, + operations: { method: 'POST', responseBodyBytes: operationResponseBodyBytes }, + session: { method: 'GET', responseBodyBytes: defaultResponseBodyBytes }, + snapshot: { method: 'GET', responseBodyBytes: snapshotResponseBodyBytes }, +} as const; + +export type ProtocolEndpoint = keyof typeof PROTOCOL_ENDPOINTS; + +export function resolveProtocolEndpoint( + value: string, +): ProtocolEndpoint | undefined { + return Object.hasOwn(PROTOCOL_ENDPOINTS, value) + ? (value as ProtocolEndpoint) + : undefined; +} + +export interface ProtocolDispatcher { + readonly sessionMetadata: SessionMetadata; + dispatch( + endpoint: ProtocolEndpoint, + body?: Uint8Array, + ): Promise; +} + +const diagnosticKeys = new Set(['message', 'reason']); +const fatalUtf8Decoder = new TextDecoder('utf-8', { fatal: true }); +const actionKey = (actions: Set) => + JSON.stringify([...actions].sort()); + +export function redactProtocolDiagnostics( + value: unknown, + redact: DiagnosticRedactor, + key?: string, +): unknown { + if (typeof value === 'string') { + if (key === undefined || !diagnosticKeys.has(key)) return value; + const redacted = redact(value); + return redacted.length > value.length + ? redacted.slice(0, value.length) + : redacted; + } + if (Array.isArray(value)) { + return value.map((entry) => redactProtocolDiagnostics(entry, redact)); + } + if (typeof value !== 'object' || value === null) return value; + return Object.fromEntries( + Object.entries(value).map(([entryKey, entry]) => [ + entryKey, + redactProtocolDiagnostics(entry, redact, entryKey), + ]), + ); +} + +export function createProtocolDispatcher( + handlers: ProtocolHandlers = {}, +): ProtocolDispatcher { + const commandRecords = new Map< + string, + { + readonly fingerprint: string; + readonly response: Promise; + } + >(); + let issuedNativeActions = new Map>(); + + return { + sessionMetadata: { + protocolVersion: PROTOCOL_VERSION, + capabilities: { + branchSearch: handlers.branchSearch !== undefined, + commands: handlers.commands !== undefined, + commitDrafts: handlers.commitDrafts !== undefined, + diff: handlers.diff !== undefined, + events: true, + nativeActions: handlers.nativeActions !== undefined, + operationRecovery: handlers.operationRecovery !== undefined, + }, + limits: PROTOCOL_LIMITS, + }, + async dispatch(endpoint, body) { + if (endpoint === 'snapshot' && handlers.snapshot !== undefined) { + const snapshot = repositorySnapshotSchema.safeParse( + await handlers.snapshot(), + ); + if (!snapshot.success) return invalidHandlerResponse(); + const nativeActions = collectNativeActions(snapshot.data); + if (nativeActions === undefined) return invalidHandlerResponse(); + issuedNativeActions = nativeActions; + return { status: 200, value: snapshot.data }; + } + if (endpoint === 'diff' && handlers.diff !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload(diffRequestSchema, input.value); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return diffResponse( + request.value.fileId, + await handlers.diff(request.value), + ); + } + if (endpoint === 'branches' && handlers.branchSearch !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + branchSearchRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return validatedResponse( + branchSearchResultSchema, + await handlers.branchSearch(request.value), + ); + } + if (endpoint === 'draft' && handlers.commitDrafts !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + commitDraftUpdateSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return commitDraftResponse( + request.value.worktreeId, + await handlers.commitDrafts(request.value), + ); + } + if (endpoint === 'commands' && handlers.commands !== undefined) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + commandEnvelopeSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + const fingerprint = fingerprintCommand(request.value.command); + const existing = commandRecords.get(request.value.clientCommandId); + if (existing !== undefined) { + if (existing.fingerprint !== fingerprint) { + return commandCollisionResponse(); + } + let response: ProtocolDispatchResponse; + try { + response = await existing.response; + } catch { + return duplicateCommandResponse(); + } + const receipt = operationReceiptSchema.safeParse(response.value); + if (response.status !== 200 || !receipt.success) { + return duplicateCommandResponse(); + } + return { + status: 200, + value: { ...receipt.data, disposition: 'duplicate' }, + }; + } + const dispatchCommand = handlers.commands; + const response = Promise.resolve() + .then(() => dispatchCommand(request.value)) + .then((value) => acceptedCommandResponse(request.value, value)); + commandRecords.set(request.value.clientCommandId, { + fingerprint, + response, + }); + return response; + } + if ( + endpoint === 'operations' && + handlers.operationRecovery !== undefined + ) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + operationRecoveryRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + return operationRecoveryResponse( + request.value.operationId, + await handlers.operationRecovery(request.value.operationId), + ); + } + if ( + endpoint === 'native-actions' && + handlers.nativeActions !== undefined + ) { + const input = parseJsonBody(body); + if (!input.ok) return input.response; + const request = parseProtocolPayload( + nativeActionRequestSchema, + input.value, + ); + if (!request.ok) { + return { status: 400, value: { error: request.error } }; + } + if ( + !issuedNativeActions + .get(request.value.targetId) + ?.has(request.value.kind) + ) { + return staleNativeTargetResponse(); + } + return validatedResponse( + nativeActionResultSchema, + await handlers.nativeActions(request.value), + ); + } + return undefined; + }, + }; +} + +function collectNativeActions( + snapshot: RepositorySnapshot, +): Map> | undefined { + const issued = new Map>(); + for (const worktree of snapshot.worktrees) { + const descriptors = [ + ...worktree.nativeTargets, + ...worktree.changes.flatMap((change) => change.nativeTargets), + ]; + for (const descriptor of descriptors) { + const actions = new Set(descriptor.actions); + const existing = issued.get(descriptor.targetId); + if (existing && actionKey(existing) !== actionKey(actions)) + return undefined; + issued.set(descriptor.targetId, actions); + } + } + return issued; +} + +function fingerprintCommand(command: CommandEnvelope['command']): string { + return createHash('sha256').update(JSON.stringify(command)).digest('hex'); +} + +function diffResponse( + fileId: FileId, + value: unknown, +): ProtocolDispatchResponse { + const result = diffResultSchema.safeParse(value); + if (!result.success || result.data.fileId !== fileId) { + return invalidHandlerResponse(); + } + return { status: 200, value: result.data }; +} + +function commitDraftResponse( + worktreeId: WorktreeId, + value: unknown, +): ProtocolDispatchResponse { + const draft = commitDraftSchema.safeParse(value); + if (!draft.success || draft.data.worktreeId !== worktreeId) { + return invalidHandlerResponse(); + } + return { status: 200, value: draft.data }; +} + +function staleNativeTargetResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'stale_target', + message: 'The native target is stale or does not allow this action.', + }, + }, + }; +} + +function operationRecoveryResponse( + operationId: OperationId, + value: unknown, +): ProtocolDispatchResponse { + const result = operationResultSchema.safeParse(value); + if (!result.success || result.data.operationId !== operationId) { + return invalidHandlerResponse(); + } + return { status: 200, value: result.data }; +} + +function acceptedCommandResponse( + request: CommandEnvelope, + value: unknown, +): ProtocolDispatchResponse { + const receipt = operationReceiptSchema.safeParse(value); + if ( + !receipt.success || + receipt.data.clientCommandId !== request.clientCommandId || + receipt.data.disposition !== 'accepted' + ) { + return invalidHandlerResponse(); + } + return { status: 200, value: receipt.data }; +} + +function commandCollisionResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'command_id_collision', + message: 'The client command ID was already used for another command.', + }, + }, + }; +} + +function duplicateCommandResponse(): ProtocolDispatchResponse { + return { + status: 409, + value: { + error: { + code: 'duplicate_command', + message: 'The duplicate command cannot be replayed safely.', + }, + }, + }; +} + +interface RuntimeSchema { + safeParse( + input: unknown, + ): { readonly data: T; readonly success: true } | { readonly success: false }; +} + +function validatedResponse( + schema: RuntimeSchema, + value: unknown, +): ProtocolDispatchResponse { + const parsed = schema.safeParse(value); + if (parsed.success) return { status: 200, value: parsed.data }; + return invalidHandlerResponse(); +} + +function invalidHandlerResponse(): ProtocolDispatchResponse { + return { + status: 500, + value: { + error: { + code: 'internal_error', + message: 'The protocol handler returned an invalid response.', + }, + }, + }; +} + +type JsonBodyResult = + | { readonly ok: true; readonly value: unknown } + | { readonly ok: false; readonly response: ProtocolDispatchResponse }; + +function parseJsonBody(body: Uint8Array | undefined): JsonBodyResult { + try { + return { + ok: true, + value: JSON.parse(fatalUtf8Decoder.decode(body)) as unknown, + }; + } catch { + return { + ok: false, + response: { + status: 400, + value: { + error: { + code: 'invalid_payload', + message: 'The protocol request body is not valid JSON.', + }, + }, + }, + }; + } +} diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index d0908eb..da8148b 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -1 +1,5 @@ export * from './loopback-server.js'; +export type { + NativeActionHandler, + ProtocolHandlers, +} from './protocol-dispatch.js'; diff --git a/packages/protocol/src/schemas.ts b/packages/protocol/src/schemas.ts index 56aeb06..764b809 100644 --- a/packages/protocol/src/schemas.ts +++ b/packages/protocol/src/schemas.ts @@ -196,10 +196,13 @@ export const changedFileSchema = z.discriminatedUnion('kind', [ }), ]); +const sanitizedRemoteHostPattern = + /^(?:\[[\da-f:.]+\]|[\da-z](?:[\da-z.-]*[\da-z])?)(?::\d{1,5})?$/iu; + export const remoteSummarySchema = z.strictObject({ remoteId: remoteIdSchema, displayName: z.string().min(1).max(256), - host: z.string().min(1).max(1_024), + host: z.string().min(1).max(1_024).regex(sanitizedRemoteHostPattern), }); export const worktreeSnapshotSchema = z.strictObject({