diff --git a/PROTOCOL.md b/PROTOCOL.md index 453b50e2..0c1dc22b 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -258,6 +258,9 @@ interface ControlHandshakeResponse { // fatal, returned by the custom handshake handler | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT'; + // Application-defined rejection data. River populates this only for + // custom-handler rejections. Older peers ignore this optional field. + extras?: unknown; }; } @@ -628,6 +631,7 @@ The server will send an error response if either: - server is in the future (`server.seq > client.nextExpectedSeq`) When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection. +Custom handshake handlers can attach application-defined `extras` to a rejection. River transports `extras` opaquely, preserves its own `code` for protocol behavior, and exposes `extras` on the client's `handshake_failed` protocol error event. Applications define and validate their own schema for `extras`, so they can make decisions (for example, stop retrying on a terminal failure) without parsing the human-readable `reason` or adding error types to River. ### Re-handshaking (live credential refresh) diff --git a/README.md b/README.md index e185fd34..ee0a9597 100644 --- a/README.md +++ b/README.md @@ -807,8 +807,9 @@ createServer(serverTransport, services, { // from?: TransportClientId, // ) => // | 'REJECTED_BY_CUSTOM_HANDLER' | 'REJECTED_UNSUPPORTED_CLIENT' (if you reject it) + // | HandshakeRejection (if you reject it with application-defined extras) // | ParsedMetadata (if you allow it) - // | a Promise of either + // | a Promise of any of the above // // next time a connection happens on the same session, previousMetadata will // be populated with the last returned value. `from` is the client id the peer @@ -820,6 +821,26 @@ createServer(serverTransport, services, { }); ``` +Use `rejectHandshake` when the client needs a machine-readable, application-defined reason for a rejection. The argument is an opaque `extras` value; your application owns its schema: + +```ts +createServerHandshakeOptions(handshakeSchema, async (metadata) => { + const authenticated = await authenticate(metadata.token); + if (!authenticated.ok) { + return rejectHandshake({ + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + }); + } + + return { parsedToken: metadata.token }; +}); +``` + +River sends `extras` on the optional `extras` field of the failed handshake response and exposes it on the client's `handshake_failed` protocol error event as `event.extras`. River does not interpret it: validate it with your own schema (for example with TypeBox's `Value.Check`) before acting on it. Existing failure codes remain available for simple handlers. Do not put secrets or raw internal errors in `extras` because River sends it to the peer. + +During a re-handshake, River exposes the rejection extras only on the server's `handshake_failed` event before it closes the session. The client's next fresh handshake can receive the extras. + `createClientHandshakeOptions` also takes an optional third `eager` argument. When set, the client constructs handshake metadata as soon as it starts dialing, so a slow `construct` (e.g. fetching a fresh token) overlaps establishing the connection instead of running after diff --git a/__tests__/e2e.test.ts b/__tests__/e2e.test.ts index be360d73..abded760 100644 --- a/__tests__/e2e.test.ts +++ b/__tests__/e2e.test.ts @@ -42,6 +42,7 @@ import { import { createClientHandshakeOptions, createServerHandshakeOptions, + rejectHandshake, } from '../router/handshake'; import { RehandshakeStreamId } from '../transport/message'; import { TestSetupHelpers } from '../testUtil/fixtures/transports'; @@ -1483,13 +1484,14 @@ describe.each(testMatrix())( 'client', createClientHandshakeOptions(requestSchema, construct), ); - const validate = vi.fn( - ( - metadata: ParsedMetadata, - ): ParsedMetadata | 'REJECTED_BY_CUSTOM_HANDLER' => - metadata.token === 'token-v1' - ? { token: metadata.token } - : 'REJECTED_BY_CUSTOM_HANDLER', + const rejectionExtras = { + code: 'TOKEN_EXPIRED', + message: 'The refreshed token expired', + }; + const validate = vi.fn((metadata: ParsedMetadata) => + metadata.token === 'token-v1' + ? { token: metadata.token } + : rejectHandshake(rejectionExtras), ); const serverTransport = getServerTransport< typeof requestSchema, @@ -1504,6 +1506,8 @@ describe.each(testMatrix())( addPostTestCleanup(async () => { await cleanupTransports([clientTransport, serverTransport]); }); + const serverHandshakeFailed = vi.fn(); + serverTransport.addEventListener('protocolError', serverHandshakeFailed); const ServiceSchema = createServiceSchema< MaybeDisposable, @@ -1539,6 +1543,12 @@ describe.each(testMatrix())( expect(serverTransport.sessions.has('client')).toBe(false), ); await waitFor(() => expect(numberOfConnections(clientTransport)).toBe(0)); + expect(serverHandshakeFailed).toHaveBeenCalledWith({ + type: 'handshake_failed', + code: 'REJECTED_BY_CUSTOM_HANDLER', + message: 're-handshake metadata rejected by handshake handler', + extras: rejectionExtras, + }); // let the client's now-disconnected session lapse before cleanup await advanceFakeTimersBySessionGrace(); diff --git a/protobuf/handshake.ts b/protobuf/handshake.ts index f441ee4c..0ede30bf 100644 --- a/protobuf/handshake.ts +++ b/protobuf/handshake.ts @@ -3,26 +3,19 @@ import type { MessageInitShape, MessageShape, } from '@bufbuild/protobuf'; -import { type Static } from 'typebox'; import { createClientHandshakeOptions as createTransportClientHandshakeOptions, createServerHandshakeOptions as createTransportServerHandshakeOptions, type ClientHandshakeOptions, + type HandshakeValidationResult, type ServerHandshakeOptions, } from '../router/handshake'; -import { - HandshakeErrorCustomHandlerFatalResponseCodes, - type TransportClientId, -} from '../transport/message'; +import { type TransportClientId } from '../transport/message'; import { decodeMessageBytes, encodeMessageBytes } from './shared'; import { Uint8ArrayType } from '../customSchemas'; const HandshakeBytesSchema = Uint8ArrayType(); -type ProtobufHandshakeFailureCode = Static< - typeof HandshakeErrorCustomHandlerFatalResponseCodes ->; - type ConstructHandshake = () => | MessageInitShape | Promise>; @@ -32,9 +25,8 @@ type ValidateHandshake = ( previousParsedMetadata?: ParsedMetadata, from?: TransportClientId, ) => - | ParsedMetadata - | ProtobufHandshakeFailureCode - | Promise; + | HandshakeValidationResult + | Promise>; /** * Create client-side handshake options backed by a protobuf message type. @@ -73,7 +65,7 @@ export function createServerHandshakeOptions< try { decoded = decodeMessageBytes(schema, metadata); } catch { - return 'REJECTED_BY_CUSTOM_HANDLER' as ProtobufHandshakeFailureCode; + return 'REJECTED_BY_CUSTOM_HANDLER'; } return await validate(decoded, previousParsedMetadata, from); diff --git a/protobuf/index.ts b/protobuf/index.ts index 53d1a8ff..5a3df140 100644 --- a/protobuf/index.ts +++ b/protobuf/index.ts @@ -21,6 +21,8 @@ export { createClientHandshakeOptions, createServerHandshakeOptions, } from './handshake'; +export { rejectHandshake } from '../router/handshake'; +export type { HandshakeRejection } from '../router/handshake'; export { createProtoService } from './service'; export type { AnyProtoService, diff --git a/router/handshake.ts b/router/handshake.ts index 427b86b6..9c65f090 100644 --- a/router/handshake.ts +++ b/router/handshake.ts @@ -4,6 +4,47 @@ import { type TransportClientId, } from '../transport/message'; +const handshakeRejectionBrand: unique symbol = Symbol('handshakeRejection'); + +export interface HandshakeRejection { + readonly [handshakeRejectionBrand]: true; + responseCode: Static; + /** + * Application-defined rejection data, forwarded to the peer in the failed + * handshake response. River transports it opaquely; the application defines + * and validates its own schema for it. + */ + extras: unknown; +} + +export function rejectHandshake( + extras: unknown, + responseCode: Static< + typeof HandshakeErrorCustomHandlerFatalResponseCodes + > = 'REJECTED_BY_CUSTOM_HANDLER', +): HandshakeRejection { + return { + [handshakeRejectionBrand]: true, + responseCode, + extras, + }; +} + +export function isHandshakeRejection( + value: unknown, +): value is HandshakeRejection { + return ( + typeof value === 'object' && + value !== null && + handshakeRejectionBrand in value + ); +} + +export type HandshakeValidationResult = + | Static + | HandshakeRejection + | ParsedMetadata; + type ConstructHandshake = () => | Static | Promise>; @@ -13,12 +54,8 @@ type ValidateHandshake = ( previousParsedMetadata?: ParsedMetadata, from?: TransportClientId, ) => - | Static - | ParsedMetadata - | Promise< - | Static - | ParsedMetadata - >; + | HandshakeValidationResult + | Promise>; export interface ClientHandshakeOptions< MetadataSchema extends TSchema = TSchema, @@ -57,8 +94,8 @@ export interface ServerHandshakeOptions< /** * Parses the metadata sent by the client during the handshake into the - * server-side {@link ParsedMetadata}, or returns a handshake failure code to - * reject the connection. + * server-side {@link ParsedMetadata}, or returns a handshake failure code or + * {@link HandshakeRejection} to reject the connection. * * @param metadata - The metadata sent by the client. * @param previousParsedMetadata - The parsed metadata from the previous diff --git a/router/index.ts b/router/index.ts index f748101b..d21ca96d 100644 --- a/router/index.ts +++ b/router/index.ts @@ -64,5 +64,7 @@ export type { export { createClientHandshakeOptions, createServerHandshakeOptions, + rejectHandshake, } from './handshake'; +export type { HandshakeRejection } from './handshake'; export { version as RIVER_VERSION } from '../package.json'; diff --git a/transport/client.ts b/transport/client.ts index 4f9e440a..177c2cee 100644 --- a/transport/client.ts +++ b/transport/client.ts @@ -377,10 +377,17 @@ export abstract class ClientTransport< ); const reason = `handshake failed: ${msg.payload.status.reason}`; + const { extras } = msg.payload.status; const to = session.to; this.rejectHandshakeResponse(session, reason, { ...session.loggingMetadata, transportMessage: msg, + ...(extras !== undefined && { + extras: { + ...session.loggingMetadata.extras, + handshakeRejectionExtras: extras, + }, + }), }); if (retriable) { @@ -390,6 +397,7 @@ export abstract class ClientTransport< type: ProtocolError.HandshakeFailed, code: msg.payload.status.code, message: reason, + ...(extras !== undefined && { extras }), }); } diff --git a/transport/events.ts b/transport/events.ts index 5da1f81e..27ed2663 100644 --- a/transport/events.ts +++ b/transport/events.ts @@ -38,6 +38,12 @@ export interface EventMap { type: (typeof ProtocolError)['HandshakeFailed']; code: Static; message: string; + /** + * Application-defined rejection data from the server's handshake + * handler, transported opaquely. Present only for custom-handler + * rejections; validate it with the application's own schema. + */ + extras?: unknown; } | { type: Omit< diff --git a/transport/message.test.ts b/transport/message.test.ts index cc2426da..f4e72b7d 100644 --- a/transport/message.test.ts +++ b/transport/message.test.ts @@ -1,6 +1,8 @@ import { TransportMessage } from '.'; import { + ControlMessageHandshakeResponseSchema, ControlFlags, + HandshakeErrorResponseCodes, handshakeRequestMessage, handshakeResponseMessage, isAck, @@ -8,6 +10,8 @@ import { isStreamOpen, } from './message'; import { describe, test, expect } from 'vitest'; +import { Type } from 'typebox'; +import { Value } from 'typebox/value'; const msg = ( to: string, @@ -105,6 +109,74 @@ describe('message helpers', () => { expect(mFail.payload.status.ok).toBe(false); }); + test('handshake rejection extras are compatible with older clients', () => { + const oldHandshakeResponseSchema = Type.Object({ + type: Type.Literal('HANDSHAKE_RESP'), + status: Type.Union([ + Type.Object({ + ok: Type.Literal(true), + sessionId: Type.String(), + }), + Type.Object({ + ok: Type.Literal(false), + reason: Type.String(), + code: HandshakeErrorResponseCodes, + }), + ]), + }); + const payload = { + type: 'HANDSHAKE_RESP', + status: { + ok: false, + reason: 'rejected by handshake handler', + code: 'REJECTED_BY_CUSTOM_HANDLER', + extras: { + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + }, + }, + }; + + expect(Value.Check(oldHandshakeResponseSchema, payload)).toBe(true); + expect(Value.Check(ControlMessageHandshakeResponseSchema, payload)).toBe( + true, + ); + }); + + test('handshake rejection extras accept any application-defined shape', () => { + for (const extras of [ + 'TERMINAL_REPL_GONE', + { code: 'TOKEN_EXPIRED' }, + 42, + null, + ]) { + expect( + Value.Check(ControlMessageHandshakeResponseSchema, { + type: 'HANDSHAKE_RESP', + status: { + ok: false, + reason: 'rejected by handshake handler', + code: 'REJECTED_BY_CUSTOM_HANDLER', + extras, + }, + }), + ).toBe(true); + } + }); + + test('handshake rejections without extras remain valid', () => { + expect( + Value.Check(ControlMessageHandshakeResponseSchema, { + type: 'HANDSHAKE_RESP', + status: { + ok: false, + reason: 'rejected by handshake handler', + code: 'REJECTED_BY_CUSTOM_HANDLER', + }, + }), + ).toBe(true); + }); + test('default message has no control flags set', () => { const m = msg('a', 'b', 'stream', { test: 1 }, 'svc', 'proc'); diff --git a/transport/message.ts b/transport/message.ts index 5dbd0cd2..a78e3ae7 100644 --- a/transport/message.ts +++ b/transport/message.ts @@ -134,6 +134,9 @@ export const ControlMessageHandshakeResponseSchema = Type.Object({ ok: Type.Literal(false), reason: Type.String(), code: HandshakeErrorResponseCodes, + // Application-defined rejection data. River populates this only for + // custom-handler rejections; older peers ignore it. + extras: Type.Optional(Type.Unknown()), }), ]), }); diff --git a/transport/server.ts b/transport/server.ts index ef42d22d..bb76c426 100644 --- a/transport/server.ts +++ b/transport/server.ts @@ -1,5 +1,8 @@ import { SpanStatusCode } from '@opentelemetry/api'; -import { ServerHandshakeOptions } from '../router/handshake'; +import { + isHandshakeRejection, + type ServerHandshakeOptions, +} from '../router/handshake'; import { validationErrorToRiverErrors } from '../router/errors'; import { ControlMessageHandshakeRequestSchema, @@ -184,9 +187,9 @@ export abstract class ServerTransport< const previousParsedMetadata = this.sessionHandshakeMetadata.get(from); - let parsedMetadataOrFailureCode; + let validationResult; try { - parsedMetadataOrFailureCode = await handshakeExtensions.validate( + validationResult = await handshakeExtensions.validate( metadata, previousParsedMetadata, from, @@ -204,10 +207,21 @@ export abstract class ServerTransport< return; } + if (isHandshakeRejection(validationResult)) { + this.teardownForFailedRehandshake( + session, + 're-handshake metadata rejected by handshake handler', + validationResult.responseCode, + validationResult.extras, + ); + + return; + } + if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, - parsedMetadataOrFailureCode, + validationResult, ) ) { this.teardownForFailedRehandshake( @@ -225,10 +239,7 @@ export abstract class ServerTransport< return; } - this.storeSessionMetadata( - session, - parsedMetadataOrFailureCode as ParsedMetadata, - ); + this.storeSessionMetadata(session, validationResult as ParsedMetadata); this.log?.info(`re-handshake from ${from} ok`, { ...session.loggingMetadata, @@ -246,6 +257,10 @@ export abstract class ServerTransport< private teardownForFailedRehandshake( session: ServerSession, reason: string, + code: Static< + typeof HandshakeErrorCustomHandlerFatalResponseCodes + > = 'REJECTED_BY_CUSTOM_HANDLER', + rejectionExtras?: unknown, ) { if (this.sessions.get(session.to) !== session) { return; @@ -255,12 +270,19 @@ export abstract class ServerTransport< this.log?.warn(`tearing down session to ${to}: ${reason}`, { ...session.loggingMetadata, connectedTo: to, + ...(rejectionExtras !== undefined && { + extras: { + ...session.loggingMetadata.extras, + handshakeRejectionExtras: rejectionExtras, + }, + }), }); this.protocolError({ type: ProtocolError.HandshakeFailed, - code: 'REJECTED_BY_CUSTOM_HANDLER', + code, message: reason, + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }); this.deleteSession(session, { unhealthy: true }); } @@ -352,13 +374,24 @@ export abstract class ServerTransport< reason: string, code: Static, metadata: MessageMetadata, + rejectionExtras?: unknown, ) { session.conn.telemetry?.span.setStatus({ code: SpanStatusCode.ERROR, message: reason, }); - this.log?.warn(reason, metadata); + const logMetadata = + rejectionExtras !== undefined + ? { + ...metadata, + extras: { + ...metadata.extras, + handshakeRejectionExtras: rejectionExtras, + }, + } + : metadata; + this.log?.warn(reason, logMetadata); const responseMsg = handshakeResponseMessage({ from: this.clientId, @@ -367,6 +400,7 @@ export abstract class ServerTransport< ok: false, code, reason, + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }, }); @@ -390,6 +424,7 @@ export abstract class ServerTransport< type: ProtocolError.HandshakeFailed, code, message: reason, + ...(rejectionExtras !== undefined && { extras: rejectionExtras }), }); this.deletePendingSession(session); } @@ -463,9 +498,9 @@ export abstract class ServerTransport< msg.from, ); - let parsedMetadataOrFailureCode; + let validationResult; try { - parsedMetadataOrFailureCode = await this.handshakeExtensions.validate( + validationResult = await this.handshakeExtensions.validate( msg.payload.metadata, previousParsedMetadata, msg.from, @@ -493,17 +528,34 @@ export abstract class ServerTransport< } // handler rejected the connection + if (isHandshakeRejection(validationResult)) { + this.rejectHandshakeRequest( + session, + msg.from, + 'rejected by handshake handler', + validationResult.responseCode, + { + ...session.loggingMetadata, + connectedTo: msg.from, + clientId: this.clientId, + }, + validationResult.extras, + ); + + return; + } + if ( Value.Check( HandshakeErrorCustomHandlerFatalResponseCodes, - parsedMetadataOrFailureCode, + validationResult, ) ) { this.rejectHandshakeRequest( session, msg.from, 'rejected by handshake handler', - parsedMetadataOrFailureCode, + validationResult, { ...session.loggingMetadata, connectedTo: msg.from, @@ -515,7 +567,7 @@ export abstract class ServerTransport< } // success! - parsedMetadata = parsedMetadataOrFailureCode as ParsedMetadata; + parsedMetadata = validationResult as ParsedMetadata; } // 4 connect cases diff --git a/transport/transport.test.ts b/transport/transport.test.ts index ba1bc0d8..e66ebf9a 100644 --- a/transport/transport.test.ts +++ b/transport/transport.test.ts @@ -29,6 +29,7 @@ import { ProvidedClientTransportOptions, ProvidedTransportOptions, } from './options'; +import { rejectHandshake } from '../router/handshake'; describe.each(testMatrix())( 'transport connection behaviour tests ($transport.name transport, $codec.name codec)', @@ -1945,5 +1946,61 @@ describe.each(testMatrix())( serverTransport, }); }); + + test('parse can reject connection with application-defined extras', async () => { + const schema = Type.Object({ foo: Type.String() }); + const rejectionExtras = { + code: 'TOKEN_EXPIRED', + message: 'The authentication token expired', + }; + const serverTransport = getServerTransport('SERVER', { + schema, + validate: async () => rejectHandshake(rejectionExtras), + }); + const clientTransport = getClientTransport('client', { + schema, + construct: async () => ({ foo: 'foo' }), + }); + const clientHandshakeFailed = vi.fn(); + clientTransport.addEventListener('protocolError', clientHandshakeFailed); + const serverRejectedConnection = vi.fn(); + serverTransport.addEventListener( + 'protocolError', + serverRejectedConnection, + ); + clientTransport.connect(serverTransport.clientId); + + addPostTestCleanup(async () => { + clientTransport.removeEventListener( + 'protocolError', + clientHandshakeFailed, + ); + serverTransport.removeEventListener( + 'protocolError', + serverRejectedConnection, + ); + await cleanupTransports([clientTransport, serverTransport]); + }); + + await waitFor(() => { + expect(clientHandshakeFailed).toHaveBeenCalledWith({ + type: ProtocolError.HandshakeFailed, + code: 'REJECTED_BY_CUSTOM_HANDLER', + message: 'handshake failed: rejected by handshake handler', + extras: rejectionExtras, + }); + expect(serverRejectedConnection).toHaveBeenCalledWith({ + type: ProtocolError.HandshakeFailed, + code: 'REJECTED_BY_CUSTOM_HANDLER', + message: 'rejected by handshake handler', + extras: rejectionExtras, + }); + }); + + await testFinishesCleanly({ + clientTransports: [clientTransport], + serverTransport, + }); + }); }, );