grpc
/
grpc-node
mirror of https://github.com/grpc/grpc-node.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

grpc-js: Implement server interceptors

This commit is contained in:
Michael Lumish 2024-01-16 13:39:07 -08:00
parent 666a374457
commit f52d1429fb
10 changed files with 1554 additions and 868 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
packages/grpc-js/src/call-interface.ts
View File

@ -37,7 +37,7 @@ export interface StatusObject {
} }
export type PartialStatusObject = Pick<StatusObject, 'code' | 'details'> & { export type PartialStatusObject = Pick<StatusObject, 'code' | 'details'> & {
metadata: Metadata | null; metadata?: Metadata | null | undefined;
}; };
export const enum WriteFlags { export const enum WriteFlags {

15
packages/grpc-js/src/index.ts
View File

@ -49,6 +49,7 @@ import {
import { Metadata, MetadataOptions, MetadataValue } from './metadata'; import { Metadata, MetadataOptions, MetadataValue } from './metadata';
import { import {
Server, Server,
ServerOptions,
UntypedHandleCall, UntypedHandleCall,
UntypedServiceImplementation, UntypedServiceImplementation,
} from './server'; } from './server';
@ -226,7 +227,7 @@ export const setLogVerbosity = (verbosity: LogVerbosity): void => {
logging.setLoggerVerbosity(verbosity); logging.setLoggerVerbosity(verbosity);
}; };
export { Server }; export { Server, ServerOptions };
export { ServerCredentials }; export { ServerCredentials };
export { KeyCertPair }; export { KeyCertPair };
@ -264,6 +265,18 @@ export { addAdminServicesToServer } from './admin';
export { ServiceConfig, LoadBalancingConfig, MethodConfig, RetryPolicy } from './service-config'; export { ServiceConfig, LoadBalancingConfig, MethodConfig, RetryPolicy } from './service-config';
export {
ServerListener,
FullServerListener,
ServerListenerBuilder,
Responder,
FullResponder,
ResponderBuilder,
ServerInterceptingCallInterface,
ServerInterceptingCall,
ServerInterceptor
} from './server-interceptors';
import * as experimental from './experimental'; import * as experimental from './experimental';
export { experimental }; export { experimental };

784
packages/grpc-js/src/server-call.ts
View File

@ -16,66 +16,17 @@
*/ */
import { EventEmitter } from 'events'; import { EventEmitter } from 'events';
import * as http2 from 'http2';
import { Duplex, Readable, Writable } from 'stream'; import { Duplex, Readable, Writable } from 'stream';
import * as zlib from 'zlib';
import { promisify } from 'util';
import { import {
Status, Status,
DEFAULT_MAX_SEND_MESSAGE_LENGTH,
DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH,
LogVerbosity,
} from './constants'; } from './constants';
import { Deserialize, Serialize } from './make-client'; import { Deserialize, Serialize } from './make-client';
import { Metadata } from './metadata'; import { Metadata } from './metadata';
import { StreamDecoder } from './stream-decoder';
import { ObjectReadable, ObjectWritable } from './object-stream'; import { ObjectReadable, ObjectWritable } from './object-stream';
import { ChannelOptions } from './channel-options';
import * as logging from './logging';
import { StatusObject, PartialStatusObject } from './call-interface'; import { StatusObject, PartialStatusObject } from './call-interface';
import { Deadline } from './deadline'; import { Deadline } from './deadline';
import { getErrorCode, getErrorMessage } from './error'; import { ServerInterceptingCallInterface } from './server-interceptors';
const TRACER_NAME = 'server_call';
const unzip = promisify(zlib.unzip);
const inflate = promisify(zlib.inflate);
function trace(text: string): void {
logging.trace(LogVerbosity.DEBUG, TRACER_NAME, text);
}
interface DeadlineUnitIndexSignature {
[name: string]: number;
}
const GRPC_ACCEPT_ENCODING_HEADER = 'grpc-accept-encoding';
const GRPC_ENCODING_HEADER = 'grpc-encoding';
const GRPC_MESSAGE_HEADER = 'grpc-message';
const GRPC_STATUS_HEADER = 'grpc-status';
const GRPC_TIMEOUT_HEADER = 'grpc-timeout';
const DEADLINE_REGEX = /(\d{1,8})\s*([HMSmun])/;
const deadlineUnitsToMs: DeadlineUnitIndexSignature = {
H: 3600000,
M: 60000,
S: 1000,
m: 1,
u: 0.001,
n: 0.000001,
};
const defaultCompressionHeaders = {
// TODO(cjihrig): Remove these encoding headers from the default response
// once compression is integrated.
[GRPC_ACCEPT_ENCODING_HEADER]: 'identity,deflate,gzip',
[GRPC_ENCODING_HEADER]: 'identity',
};
const defaultResponseHeaders = {
[http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
[http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
};
const defaultResponseOptions = {
waitForTrailers: true,
} as http2.ServerStreamResponseOptions;
export type ServerStatusResponse = Partial<StatusObject>; export type ServerStatusResponse = Partial<StatusObject>;
@ -105,6 +56,38 @@ export type ServerDuplexStream<RequestType, ResponseType> = ServerSurfaceCall &
ObjectReadable<RequestType> & ObjectReadable<RequestType> &
ObjectWritable<ResponseType> & { end: (metadata?: Metadata) => void }; ObjectWritable<ResponseType> & { end: (metadata?: Metadata) => void };
export function serverErrorToStatus(error: ServerErrorResponse | ServerStatusResponse, extraTrailers?: Metadata | undefined): PartialStatusObject {
const status: PartialStatusObject = {
code: Status.UNKNOWN,
details: 'message' in error ? error.message : 'Unknown Error',
metadata:
'metadata' in error && error.metadata !== undefined
? error.metadata
: null,
};
if (
'code' in error &&
typeof error.code === 'number' &&
Number.isInteger(error.code)
) {
status.code = error.code;
if ('details' in error && typeof error.details === 'string') {
status.details = error.details!;
}
}
if (extraTrailers) {
if (status.metadata) {
status.metadata.merge(extraTrailers);
} else {
status.metadata = extraTrailers;
}
}
return status;
}
export class ServerUnaryCallImpl<RequestType, ResponseType> export class ServerUnaryCallImpl<RequestType, ResponseType>
extends EventEmitter extends EventEmitter
implements ServerUnaryCall<RequestType, ResponseType> implements ServerUnaryCall<RequestType, ResponseType>
@ -112,13 +95,13 @@ export class ServerUnaryCallImpl<RequestType, ResponseType>
cancelled: boolean; cancelled: boolean;
constructor( constructor(
private call: Http2ServerCallStream<RequestType, ResponseType>, private path: string,
private call: ServerInterceptingCallInterface,
public metadata: Metadata, public metadata: Metadata,
public request: RequestType public request: RequestType
) { ) {
super(); super();
this.cancelled = false; this.cancelled = false;
this.call.setupSurfaceCall(this);
} }
getPeer(): string { getPeer(): string {
@ -134,7 +117,7 @@ export class ServerUnaryCallImpl<RequestType, ResponseType>
} }
getPath(): string { getPath(): string {
return this.call.getPath(); return this.path;
} }
} }
@ -145,23 +128,16 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
cancelled: boolean; cancelled: boolean;
constructor( constructor(
private call: Http2ServerCallStream<RequestType, ResponseType>, private path: string,
public metadata: Metadata, private call: ServerInterceptingCallInterface,
public deserialize: Deserialize<RequestType>, public metadata: Metadata
encoding: string
) { ) {
super({ objectMode: true }); super({ objectMode: true });
this.cancelled = false; this.cancelled = false;
this.call.setupSurfaceCall(this);
this.call.setupReadable(this, encoding);
} }
_read(size: number) { _read(size: number) {
if (!this.call.consumeUnpushedMessages(this)) { this.call.startRead();
return;
}
this.call.resume();
} }
getPeer(): string { getPeer(): string {
@ -177,7 +153,7 @@ export class ServerReadableStreamImpl<RequestType, ResponseType>
} }
getPath(): string { getPath(): string {
return this.call.getPath(); return this.path;
} }
} }
@ -187,20 +163,23 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
{ {
cancelled: boolean; cancelled: boolean;
private trailingMetadata: Metadata; private trailingMetadata: Metadata;
private pendingStatus: PartialStatusObject = {
code: Status.OK,
details: 'OK'
};
constructor( constructor(
private call: Http2ServerCallStream<RequestType, ResponseType>, private path: string,
private call: ServerInterceptingCallInterface,
public metadata: Metadata, public metadata: Metadata,
public serialize: Serialize<ResponseType>,
public request: RequestType public request: RequestType
) { ) {
super({ objectMode: true }); super({ objectMode: true });
this.cancelled = false; this.cancelled = false;
this.trailingMetadata = new Metadata(); this.trailingMetadata = new Metadata();
this.call.setupSurfaceCall(this);
this.on('error', err => { this.on('error', err => {
this.call.sendError(err); this.pendingStatus = serverErrorToStatus(err);
this.end(); this.end();
}); });
} }
@ -218,7 +197,7 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
} }
getPath(): string { getPath(): string {
return this.call.getPath(); return this.path;
} }
_write( _write(
@ -227,28 +206,13 @@ export class ServerWritableStreamImpl<RequestType, ResponseType>
// eslint-disable-next-line @typescript-eslint/no-explicit-any // eslint-disable-next-line @typescript-eslint/no-explicit-any
callback: (...args: any[]) => void callback: (...args: any[]) => void
) { ) {
try { this.call.sendMessage(chunk, callback);
const response = this.call.serializeMessage(chunk);
if (!this.call.write(response)) {
this.call.once('drain', callback);
return;
}
} catch (err) {
this.emit('error', {
details: getErrorMessage(err),
code: Status.INTERNAL,
});
}
callback();
} }
_final(callback: Function): void { _final(callback: Function): void {
this.call.sendStatus({ this.call.sendStatus({
code: Status.OK, ...this.pendingStatus,
details: 'OK', metadata: this.pendingStatus.metadata ?? this.trailingMetadata,
metadata: this.trailingMetadata,
}); });
callback(null); callback(null);
} }
@ -268,27 +232,23 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
implements ServerDuplexStream<RequestType, ResponseType> implements ServerDuplexStream<RequestType, ResponseType>
{ {
cancelled: boolean; cancelled: boolean;
/* This field appears to be unsued, but it is actually used in _final, which is assiged from
* ServerWritableStreamImpl.prototype._final below. */
// eslint-disable-next-line @typescript-eslint/ban-ts-comment
// @ts-ignore noUnusedLocals
private trailingMetadata: Metadata; private trailingMetadata: Metadata;
private pendingStatus: PartialStatusObject = {
code: Status.OK,
details: 'OK'
};
constructor( constructor(
private call: Http2ServerCallStream<RequestType, ResponseType>, private path: string,
public metadata: Metadata, private call: ServerInterceptingCallInterface,
public serialize: Serialize<ResponseType>, public metadata: Metadata
public deserialize: Deserialize<RequestType>,
encoding: string
) { ) {
super({ objectMode: true }); super({ objectMode: true });
this.cancelled = false; this.cancelled = false;
this.trailingMetadata = new Metadata(); this.trailingMetadata = new Metadata();
this.call.setupSurfaceCall(this);
this.call.setupReadable(this, encoding);
this.on('error', err => { this.on('error', err => {
this.call.sendError(err); this.pendingStatus = serverErrorToStatus(err);
this.end(); this.end();
}); });
} }
@ -306,7 +266,28 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
} }
getPath(): string { getPath(): string {
return this.call.getPath(); return this.path;
}
_read(size: number) {
this.call.startRead();
}
_write(
chunk: ResponseType,
encoding: string,
// eslint-disable-next-line @typescript-eslint/no-explicit-any
callback: (...args: any[]) => void
) {
this.call.sendMessage(chunk, callback);
}
_final(callback: Function): void {
this.call.sendStatus({
...this.pendingStatus,
metadata: this.pendingStatus.metadata ?? this.trailingMetadata,
});
callback(null);
} }
// eslint-disable-next-line @typescript-eslint/no-explicit-any // eslint-disable-next-line @typescript-eslint/no-explicit-any
@ -319,13 +300,6 @@ export class ServerDuplexStreamImpl<RequestType, ResponseType>
} }
} }
ServerDuplexStreamImpl.prototype._read =
ServerReadableStreamImpl.prototype._read;
ServerDuplexStreamImpl.prototype._write =
ServerWritableStreamImpl.prototype._write;
ServerDuplexStreamImpl.prototype._final =
ServerWritableStreamImpl.prototype._final;
// Unary response callback signature. // Unary response callback signature.
export type sendUnaryData<ResponseType> = ( export type sendUnaryData<ResponseType> = (
error: ServerErrorResponse | ServerStatusResponse | null, error: ServerErrorResponse | ServerStatusResponse | null,
@ -401,597 +375,3 @@ export type Handler<RequestType, ResponseType> =
| BidiStreamingHandler<RequestType, ResponseType>; | BidiStreamingHandler<RequestType, ResponseType>;
export type HandlerType = 'bidi' | 'clientStream' | 'serverStream' | 'unary'; export type HandlerType = 'bidi' | 'clientStream' | 'serverStream' | 'unary';
// Internal class that wraps the HTTP2 request.
export class Http2ServerCallStream<
RequestType,
ResponseType
> extends EventEmitter {
cancelled = false;
deadlineTimer: NodeJS.Timeout | null = null;
private statusSent = false;
private deadline: Deadline = Infinity;
private wantTrailers = false;
private metadataSent = false;
private canPush = false;
private isPushPending = false;
private bufferedMessages: Array<Buffer | null> = [];
private messagesToPush: Array<RequestType | null> = [];
private maxSendMessageSize: number = DEFAULT_MAX_SEND_MESSAGE_LENGTH;
private maxReceiveMessageSize: number = DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH;
constructor(
private stream: http2.ServerHttp2Stream,
private handler: Handler<RequestType, ResponseType>,
options: ChannelOptions
) {
super();
this.stream.once('error', (err: ServerErrorResponse) => {
/* We need an error handler to avoid uncaught error event exceptions, but
* there is nothing we can reasonably do here. Any error event should
* have a corresponding close event, which handles emitting the cancelled
* event. And the stream is now in a bad state, so we can't reasonably
* expect to be able to send an error over it. */
});
this.stream.once('close', () => {
trace(
'Request to method ' +
this.handler?.path +
' stream closed with rstCode ' +
this.stream.rstCode
);
if (!this.statusSent) {
this.cancelled = true;
this.emit('cancelled', 'cancelled');
this.emit('streamEnd', false);
this.sendStatus({
code: Status.CANCELLED,
details: 'Cancelled by client',
metadata: null,
});
if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
}
});
this.stream.on('drain', () => {
this.emit('drain');
});
if ('grpc.max_send_message_length' in options) {
this.maxSendMessageSize = options['grpc.max_send_message_length']!;
}
if ('grpc.max_receive_message_length' in options) {
this.maxReceiveMessageSize = options['grpc.max_receive_message_length']!;
}
}
private checkCancelled(): boolean {
/* In some cases the stream can become destroyed before the close event
* fires. That creates a race condition that this check works around */
if (this.stream.destroyed || this.stream.closed) {
this.cancelled = true;
}
return this.cancelled;
}
private getDecompressedMessage(
message: Buffer,
encoding: string
): Buffer | Promise<Buffer> {
if (encoding === 'deflate') {
return inflate(message.subarray(5));
} else if (encoding === 'gzip') {
return unzip(message.subarray(5));
} else if (encoding === 'identity') {
return message.subarray(5);
}
return Promise.reject({
code: Status.UNIMPLEMENTED,
details: `Received message compressed with unsupported encoding "${encoding}"`,
});
}
sendMetadata(customMetadata?: Metadata) {
if (this.checkCancelled()) {
return;
}
if (this.metadataSent) {
return;
}
this.metadataSent = true;
const custom = customMetadata ? customMetadata.toHttp2Headers() : null;
// TODO(cjihrig): Include compression headers.
const headers = {
...defaultResponseHeaders,
...defaultCompressionHeaders,
...custom,
};
this.stream.respond(headers, defaultResponseOptions);
}
receiveMetadata(headers: http2.IncomingHttpHeaders) {
const metadata = Metadata.fromHttp2Headers(headers);
if (logging.isTracerEnabled(TRACER_NAME)) {
trace(
'Request to ' +
this.handler.path +
' received headers ' +
JSON.stringify(metadata.toJSON())
);
}
// TODO(cjihrig): Receive compression metadata.
const timeoutHeader = metadata.get(GRPC_TIMEOUT_HEADER);
if (timeoutHeader.length > 0) {
const match = timeoutHeader[0].toString().match(DEADLINE_REGEX);
if (match === null) {
const err = new Error('Invalid deadline') as ServerErrorResponse;
err.code = Status.OUT_OF_RANGE;
this.sendError(err);
return metadata;
}
const timeout = (+match[1] * deadlineUnitsToMs[match[2]]) | 0;
const now = new Date();
this.deadline = now.setMilliseconds(now.getMilliseconds() + timeout);
this.deadlineTimer = setTimeout(handleExpiredDeadline, timeout, this);
metadata.remove(GRPC_TIMEOUT_HEADER);
}
// Remove several headers that should not be propagated to the application
metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
metadata.remove(http2.constants.HTTP2_HEADER_TE);
metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
metadata.remove('grpc-accept-encoding');
return metadata;
}
receiveUnaryMessage(encoding: string): Promise<RequestType> {
return new Promise((resolve, reject) => {
const { stream } = this;
let receivedLength = 0;
// eslint-disable-next-line @typescript-eslint/no-this-alias
const call = this;
const body: Buffer[] = [];
const limit = this.maxReceiveMessageSize;
this.stream.on('data', onData);
this.stream.on('end', onEnd);
this.stream.on('error', onEnd);
function onData(chunk: Buffer) {
receivedLength += chunk.byteLength;
if (limit !== -1 && receivedLength > limit) {
stream.removeListener('data', onData);
stream.removeListener('end', onEnd);
stream.removeListener('error', onEnd);
reject({
code: Status.RESOURCE_EXHAUSTED,
details: `Received message larger than max (${receivedLength} vs. ${limit})`,
});
return;
}
body.push(chunk);
}
function onEnd(err?: Error) {
stream.removeListener('data', onData);
stream.removeListener('end', onEnd);
stream.removeListener('error', onEnd);
if (err !== undefined) {
reject({ code: Status.INTERNAL, details: err.message });
return;
}
if (receivedLength === 0) {
reject({
code: Status.INTERNAL,
details: 'received empty unary message',
});
return;
}
call.emit('receiveMessage');
const requestBytes = Buffer.concat(body, receivedLength);
const compressed = requestBytes.readUInt8(0) === 1;
const compressedMessageEncoding = compressed ? encoding : 'identity';
const decompressedMessage = call.getDecompressedMessage(
requestBytes,
compressedMessageEncoding
);
if (Buffer.isBuffer(decompressedMessage)) {
resolve(
call.deserializeMessageWithInternalError(decompressedMessage)
);
return;
}
decompressedMessage.then(
decompressed =>
resolve(call.deserializeMessageWithInternalError(decompressed)),
(err: any) =>
reject(
err.code
? err
: {
code: Status.INTERNAL,
details: `Received "grpc-encoding" header "${encoding}" but ${encoding} decompression failed`,
}
)
);
}
});
}
private async deserializeMessageWithInternalError(buffer: Buffer) {
try {
return this.deserializeMessage(buffer);
} catch (err) {
throw {
details: getErrorMessage(err),
code: Status.INTERNAL,
};
}
}
serializeMessage(value: ResponseType) {
const messageBuffer = this.handler.serialize(value);
// TODO(cjihrig): Call compression aware serializeMessage().
const byteLength = messageBuffer.byteLength;
const output = Buffer.allocUnsafe(byteLength + 5);
output.writeUInt8(0, 0);
output.writeUInt32BE(byteLength, 1);
messageBuffer.copy(output, 5);
return output;
}
deserializeMessage(bytes: Buffer) {
return this.handler.deserialize(bytes);
}
async sendUnaryMessage(
err: ServerErrorResponse | ServerStatusResponse | null,
value?: ResponseType | null,
metadata?: Metadata | null,
flags?: number
) {
if (this.checkCancelled()) {
return;
}
if (metadata === undefined) {
metadata = null;
}
if (err) {
if (!Object.prototype.hasOwnProperty.call(err, 'metadata') && metadata) {
err.metadata = metadata;
}
this.sendError(err);
return;
}
try {
const response = this.serializeMessage(value!);
this.write(response);
this.sendStatus({ code: Status.OK, details: 'OK', metadata });
} catch (err) {
this.sendError({
details: getErrorMessage(err),
code: Status.INTERNAL,
});
}
}
sendStatus(statusObj: PartialStatusObject) {
this.emit('callEnd', statusObj.code);
this.emit('streamEnd', statusObj.code === Status.OK);
if (this.checkCancelled()) {
return;
}
trace(
'Request to method ' +
this.handler?.path +
' ended with status code: ' +
Status[statusObj.code] +
' details: ' +
statusObj.details
);
if (this.deadlineTimer) clearTimeout(this.deadlineTimer);
if (this.stream.headersSent) {
if (!this.wantTrailers) {
this.wantTrailers = true;
this.stream.once('wantTrailers', () => {
const trailersToSend = {
[GRPC_STATUS_HEADER]: statusObj.code,
[GRPC_MESSAGE_HEADER]: encodeURI(statusObj.details),
...statusObj.metadata?.toHttp2Headers(),
};
this.stream.sendTrailers(trailersToSend);
this.statusSent = true;
});
this.stream.end();
}
} else {
// Trailers-only response
const trailersToSend = {
[GRPC_STATUS_HEADER]: statusObj.code,
[GRPC_MESSAGE_HEADER]: encodeURI(statusObj.details),
...defaultResponseHeaders,
...statusObj.metadata?.toHttp2Headers(),
};
this.stream.respond(trailersToSend, { endStream: true });
this.statusSent = true;
}
}
sendError(error: ServerErrorResponse | ServerStatusResponse) {
const status: PartialStatusObject = {
code: Status.UNKNOWN,
details: 'message' in error ? error.message : 'Unknown Error',
metadata:
'metadata' in error && error.metadata !== undefined
? error.metadata
: null,
};
if (
'code' in error &&
typeof error.code === 'number' &&
Number.isInteger(error.code)
) {
status.code = error.code;
if ('details' in error && typeof error.details === 'string') {
status.details = error.details!;
}
}
this.sendStatus(status);
}
write(chunk: Buffer) {
if (this.checkCancelled()) {
return;
}
if (
this.maxSendMessageSize !== -1 &&
chunk.length > this.maxSendMessageSize
) {
this.sendError({
code: Status.RESOURCE_EXHAUSTED,
details: `Sent message larger than max (${chunk.length} vs. ${this.maxSendMessageSize})`,
});
return;
}
this.sendMetadata();
this.emit('sendMessage');
return this.stream.write(chunk);
}
resume() {
this.stream.resume();
}
setupSurfaceCall(call: ServerSurfaceCall) {
this.once('cancelled', reason => {
call.cancelled = true;
call.emit('cancelled', reason);
});
this.once('callEnd', status => call.emit('callEnd', status));
}
setupReadable(
readable:
| ServerReadableStream<RequestType, ResponseType>
| ServerDuplexStream<RequestType, ResponseType>,
encoding: string
) {
const decoder = new StreamDecoder();
let readsDone = false;
let pendingMessageProcessing = false;
let pushedEnd = false;
const maybePushEnd = async () => {
if (!pushedEnd && readsDone && !pendingMessageProcessing) {
pushedEnd = true;
await this.pushOrBufferMessage(readable, null);
}
};
this.stream.on('data', async (data: Buffer) => {
const messages = decoder.write(data);
pendingMessageProcessing = true;
this.stream.pause();
for (const message of messages) {
if (
this.maxReceiveMessageSize !== -1 &&
message.length > this.maxReceiveMessageSize
) {
this.sendError({
code: Status.RESOURCE_EXHAUSTED,
details: `Received message larger than max (${message.length} vs. ${this.maxReceiveMessageSize})`,
});
return;
}
this.emit('receiveMessage');
const compressed = message.readUInt8(0) === 1;
const compressedMessageEncoding = compressed ? encoding : 'identity';
const decompressedMessage = await this.getDecompressedMessage(
message,
compressedMessageEncoding
);
// Encountered an error with decompression; it'll already have been propogated back
// Just return early
if (!decompressedMessage) return;
await this.pushOrBufferMessage(readable, decompressedMessage);
}
pendingMessageProcessing = false;
this.stream.resume();
await maybePushEnd();
});
this.stream.once('end', async () => {
readsDone = true;
await maybePushEnd();
});
}
consumeUnpushedMessages(
readable:
| ServerReadableStream<RequestType, ResponseType>
| ServerDuplexStream<RequestType, ResponseType>
): boolean {
this.canPush = true;
while (this.messagesToPush.length > 0) {
const nextMessage = this.messagesToPush.shift();
const canPush = readable.push(nextMessage);
if (nextMessage === null || canPush === false) {
this.canPush = false;
break;
}
}
return this.canPush;
}
private async pushOrBufferMessage(
readable:
| ServerReadableStream<RequestType, ResponseType>
| ServerDuplexStream<RequestType, ResponseType>,
messageBytes: Buffer | null
): Promise<void> {
if (this.isPushPending) {
this.bufferedMessages.push(messageBytes);
} else {
await this.pushMessage(readable, messageBytes);
}
}
private async pushMessage(
readable:
| ServerReadableStream<RequestType, ResponseType>
| ServerDuplexStream<RequestType, ResponseType>,
messageBytes: Buffer | null
) {
if (messageBytes === null) {
trace('Received end of stream');
if (this.canPush) {
readable.push(null);
} else {
this.messagesToPush.push(null);
}
return;
}
trace('Received message of length ' + messageBytes.length);
this.isPushPending = true;
try {
const deserialized = await this.deserializeMessage(messageBytes);
if (this.canPush) {
if (!readable.push(deserialized)) {
this.canPush = false;
this.stream.pause();
}
} else {
this.messagesToPush.push(deserialized);
}
} catch (error) {
// Ignore any remaining messages when errors occur.
this.bufferedMessages.length = 0;
let code = getErrorCode(error);
if (code === null || code < Status.OK || code > Status.UNAUTHENTICATED) {
code = Status.INTERNAL;
}
readable.emit('error', {
details: getErrorMessage(error),
code: code,
});
}
this.isPushPending = false;
if (this.bufferedMessages.length > 0) {
await this.pushMessage(
readable,
this.bufferedMessages.shift() as Buffer | null
);
}
}
getPeer(): string {
const socket = this.stream.session?.socket;
if (socket?.remoteAddress) {
if (socket.remotePort) {
return `${socket.remoteAddress}:${socket.remotePort}`;
} else {
return socket.remoteAddress;
}
} else {
return 'unknown';
}
}
getDeadline(): Deadline {
return this.deadline;
}
getPath(): string {
return this.handler.path;
}
}
/* eslint-disable @typescript-eslint/no-explicit-any */
type UntypedServerCall = Http2ServerCallStream<any, any>;
function handleExpiredDeadline(call: UntypedServerCall) {
const err = new Error('Deadline exceeded') as ServerErrorResponse;
err.code = Status.DEADLINE_EXCEEDED;
call.sendError(err);
call.cancelled = true;
call.emit('cancelled', 'deadline');
}

886
packages/grpc-js/src/server-interceptors.ts Normal file
View File

@ -0,0 +1,886 @@
/*
* Copyright 2024 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
import { PartialStatusObject} from "./call-interface";
import { ServerMethodDefinition } from "./make-client";
import { Metadata } from "./metadata";
import { ChannelOptions } from "./channel-options";
import { Handler, ServerErrorResponse } from "./server-call";
import { Deadline } from "./deadline";
import { DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH, DEFAULT_MAX_SEND_MESSAGE_LENGTH, LogVerbosity, Status } from "./constants";
import * as http2 from 'http2';
import { getErrorMessage } from "./error";
import * as zlib from 'zlib';
import { promisify } from "util";
import { StreamDecoder } from "./stream-decoder";
import { CallEventTracker } from "./transport";
import * as logging from './logging';
const unzip = promisify(zlib.unzip);
const inflate = promisify(zlib.inflate);
const TRACER_NAME = 'server_call';
function trace(text: string) {
logging.trace(LogVerbosity.DEBUG, TRACER_NAME, text);
}
export interface ServerMetadataListener {
(metadata: Metadata, next: (metadata: Metadata) => void): void;
}
export interface ServerMessageListener {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(message: any, next: (message: any) => void): void;
}
export interface ServerHalfCloseListener {
(next: () => void): void;
}
export interface ServerCancelListener {
(): void;
}
export interface FullServerListener {
onReceiveMetadata: ServerMetadataListener;
onReceiveMessage: ServerMessageListener;
onReceiveHalfClose: ServerHalfCloseListener;
onCancel: ServerCancelListener;
}
export type ServerListener = Partial<FullServerListener>;
export class ServerListenerBuilder {
private metadata: ServerMetadataListener | undefined = undefined;
private message: ServerMessageListener | undefined = undefined;
private halfClose: ServerHalfCloseListener | undefined = undefined;
private cancel: ServerCancelListener | undefined = undefined;
withOnReceiveMetadata(onReceiveMetadata: ServerMetadataListener): this {
this.metadata = onReceiveMetadata;
return this;
}
withOnReceiveMessage(onReceiveMessage: ServerMessageListener): this {
this.message = onReceiveMessage;
return this;
}
withOnReceiveHalfClose(onReceiveHalfClose: ServerHalfCloseListener): this {
this.halfClose = onReceiveHalfClose;
return this;
}
withOnCancel(onCancel: ServerCancelListener): this {
this.cancel = onCancel;
return this;
}
build(): ServerListener {
return {
onReceiveMetadata: this.metadata,
onReceiveMessage: this.message,
onReceiveHalfClose: this.halfClose,
onCancel: this.cancel
};
}
}
export interface InterceptingServerListener {
onReceiveMetadata(metadata: Metadata): void;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
onReceiveMessage(message: any): void;
onReceiveHalfClose(): void;
onCancel(): void;
}
export function isInterceptingServerListener(listener: ServerListener | InterceptingServerListener): listener is InterceptingServerListener {
return listener.onReceiveMetadata !== undefined && listener.onReceiveMetadata.length === 1;
}
class InterceptingServerListenerImpl implements InterceptingServerListener {
/**
* Once the call is cancelled, ignore all other events.
*/
private cancelled: boolean = false;
private processingMetadata: boolean = false;
private hasPendingMessage: boolean = false;
private pendingMessage: any = null;
private processingMessage: boolean = false;
private hasPendingHalfClose: boolean = false;
constructor(private listener: FullServerListener, private nextListener: InterceptingServerListener) {}
private processPendingMessage() {
if (this.hasPendingMessage) {
this.nextListener.onReceiveMessage(this.pendingMessage);
this.pendingMessage = null;
this.hasPendingMessage = false;
}
}
private processPendingHalfClose() {
if (this.hasPendingHalfClose) {
this.nextListener.onReceiveHalfClose();
this.hasPendingHalfClose = false;
}
}
onReceiveMetadata(metadata: Metadata): void {
if (this.cancelled) {
return;
}
this.processingMetadata = true;
this.listener.onReceiveMetadata(metadata, interceptedMetadata => {
this.processingMetadata = false;
if (this.cancelled) {
return;
}
this.nextListener.onReceiveMetadata(interceptedMetadata);
this.processPendingMessage();
this.processPendingHalfClose();
});
}
onReceiveMessage(message: any): void {
if (this.cancelled) {
return;
}
this.processingMessage = true;
this.listener.onReceiveMessage(message, msg => {
this.processingMessage = false;
if (this.cancelled) {
return;
}
if (this.processingMetadata) {
this.pendingMessage = msg;
this.hasPendingMessage = true;
} else {
this.nextListener.onReceiveMessage(msg);
this.processPendingHalfClose();
}
});
}
onReceiveHalfClose(): void {
if (this.cancelled) {
return;
}
this.listener.onReceiveHalfClose(() => {
if (this.cancelled) {
return;
}
if (this.processingMetadata || this.processingMessage) {
this.hasPendingHalfClose = true;
} else {
this.nextListener.onReceiveHalfClose();
}
});
}
onCancel(): void {
this.cancelled = true;
this.listener.onCancel();
this.nextListener.onCancel();
}
}
export interface StartResponder {
(next: (listener?: ServerListener) => void): void;
}
export interface MetadataResponder {
(metadata: Metadata, next: (metadata: Metadata) => void): void;
}
export interface MessageResponder {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
(message: any, next: (message: any) => void): void;
}
export interface StatusResponder {
(status: PartialStatusObject, next: (status: PartialStatusObject) => void): void;
}
export interface FullResponder {
start: StartResponder;
sendMetadata: MetadataResponder;
sendMessage: MessageResponder;
sendStatus: StatusResponder;
}
export type Responder = Partial<FullResponder>;
export class ResponderBuilder {
private start: StartResponder | undefined = undefined;
private metadata: MetadataResponder | undefined = undefined;
private message: MessageResponder | undefined = undefined;
private status: StatusResponder | undefined = undefined;
withStart(start: StartResponder): this {
this.start = start;
return this;
}
withSendMetadata(sendMetadata: MetadataResponder): this {
this.metadata = sendMetadata;
return this;
}
withSendMessage(sendMessage: MessageResponder): this {
this.message = sendMessage;
return this;
}
withSendStatus(sendStatus: StatusResponder): this {
this.status = sendStatus;
return this;
}
build(): Responder {
return {
start: this.start,
sendMetadata: this.metadata,
sendMessage: this.message,
sendStatus: this.status
};
}
}
const defaultServerListener: FullServerListener = {
onReceiveMetadata: (metadata, next) => {
next(metadata);
},
onReceiveMessage: (message, next) => {
next(message);
},
onReceiveHalfClose: next => {
next();
},
onCancel: () => {}
};
const defaultResponder: FullResponder = {
start: (next) => {
next();
},
sendMetadata: (metadata, next) => {
next(metadata);
},
sendMessage: (message, next) => {
next(message);
},
sendStatus: (status, next) => {
next(status);
}
};
export interface ServerInterceptingCallInterface {
/**
* Register the listener to handle inbound events.
*/
start(listener: InterceptingServerListener): void;
/**
* Send response metadata.
*/
sendMetadata(metadata: Metadata): void;
/**
* Send a response message.
*/
sendMessage(message: any, callback: () => void): void;
/**
* End the call by sending this status.
*/
sendStatus(status: PartialStatusObject): void;
/**
* Start a single read, eventually triggering either listener.onReceiveMessage or listener.onReceiveHalfClose.
*/
startRead(): void;
/**
* Return the peer address of the client making the request, if known, or "unknown" otherwise
*/
getPeer(): string;
/**
* Return the call deadline set by the client. The value is Infinity if there is no deadline.
*/
getDeadline(): Deadline;
}
export class ServerInterceptingCall implements ServerInterceptingCallInterface {
private responder: FullResponder;
private processingMetadata: boolean = false;
private processingMessage: boolean = false;
private pendingMessage: any = null;
private pendingMessageCallback: (() => void) | null = null;
private pendingStatus: PartialStatusObject | null = null;
constructor(private nextCall: ServerInterceptingCallInterface, responder?: Responder) {
this.responder = {...defaultResponder, ...responder};
}
private processPendingMessage() {
if (this.pendingMessageCallback) {
this.nextCall.sendMessage(this.pendingMessage, this.pendingMessageCallback);
this.pendingMessage = null;
this.pendingMessageCallback = null;
}
}
private processPendingStatus() {
if (this.pendingStatus) {
this.nextCall.sendStatus(this.pendingStatus);
this.pendingStatus = null;
}
}
start(listener: InterceptingServerListener): void {
this.responder.start(interceptedListener => {
const fullInterceptedListener: FullServerListener = {...defaultServerListener, ...interceptedListener};
const finalInterceptingListener = new InterceptingServerListenerImpl(fullInterceptedListener, listener);
this.nextCall.start(finalInterceptingListener);
});
}
sendMetadata(metadata: Metadata): void {
this.processingMetadata = true;
this.responder.sendMetadata(metadata, interceptedMetadata => {
this.processingMetadata = false;
this.nextCall.sendMetadata(interceptedMetadata);
this.processPendingMessage();
this.processPendingStatus();
});
}
sendMessage(message: any, callback: () => void): void {
this.processingMessage = true;
this.responder.sendMessage(message, interceptedMessage => {
this.processingMessage = false;
if (this.processingMetadata) {
this.pendingMessage = interceptedMessage;
this.pendingMessageCallback = callback;
} else {
this.nextCall.sendMessage(interceptedMessage, callback);
}
});
}
sendStatus(status: PartialStatusObject): void {
this.responder.sendStatus(status, interceptedStatus => {
if (this.processingMetadata || this.processingMessage) {
this.pendingStatus = interceptedStatus;
} else {
this.nextCall.sendStatus(interceptedStatus);
}
});
}
startRead(): void {
this.nextCall.startRead();
}
getPeer(): string {
return this.nextCall.getPeer();
}
getDeadline(): Deadline {
return this.nextCall.getDeadline();
}
}
export interface ServerInterceptor {
(methodDescriptor: ServerMethodDefinition<any, any>, call: ServerInterceptingCallInterface): ServerInterceptingCall;
}
interface DeadlineUnitIndexSignature {
[name: string]: number;
}
const GRPC_ACCEPT_ENCODING_HEADER = 'grpc-accept-encoding';
const GRPC_ENCODING_HEADER = 'grpc-encoding';
const GRPC_MESSAGE_HEADER = 'grpc-message';
const GRPC_STATUS_HEADER = 'grpc-status';
const GRPC_TIMEOUT_HEADER = 'grpc-timeout';
const DEADLINE_REGEX = /(\d{1,8})\s*([HMSmun])/;
const deadlineUnitsToMs: DeadlineUnitIndexSignature = {
H: 3600000,
M: 60000,
S: 1000,
m: 1,
u: 0.001,
n: 0.000001,
};
const defaultCompressionHeaders = {
// TODO(cjihrig): Remove these encoding headers from the default response
// once compression is integrated.
[GRPC_ACCEPT_ENCODING_HEADER]: 'identity,deflate,gzip',
[GRPC_ENCODING_HEADER]: 'identity',
};
const defaultResponseHeaders = {
[http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
[http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
};
const defaultResponseOptions = {
waitForTrailers: true,
} as http2.ServerStreamResponseOptions;
type ReadQueueEntryType = 'COMPRESSED' | 'READABLE' | 'HALF_CLOSE';
interface ReadQueueEntry {
type: ReadQueueEntryType;
compressedMessage: Buffer | null;
parsedMessage: any;
}
export class BaseServerInterceptingCall implements ServerInterceptingCallInterface {
private listener: InterceptingServerListener | null = null;
private metadata: Metadata;
private deadlineTimer: NodeJS.Timeout | null = null;
private deadline: Deadline = Infinity;
private maxSendMessageSize: number = DEFAULT_MAX_SEND_MESSAGE_LENGTH;
private maxReceiveMessageSize: number = DEFAULT_MAX_RECEIVE_MESSAGE_LENGTH;
private cancelled = false;
private metadataSent = false;
private wantTrailers = false;
private cancelNotified = false;
private incomingEncoding: string = 'identity';
private decoder = new StreamDecoder();
private readQueue: ReadQueueEntry[] = [];
private isReadPending = false;
private receivedHalfClose = false;
private streamEnded = false;
constructor(
private readonly stream: http2.ServerHttp2Stream,
headers: http2.IncomingHttpHeaders,
private readonly callEventTracker: CallEventTracker | null,
private readonly handler: Handler<any, any>,
options: ChannelOptions
) {
this.stream.once('error', (err: ServerErrorResponse) => {
/* We need an error handler to avoid uncaught error event exceptions, but
* there is nothing we can reasonably do here. Any error event should
* have a corresponding close event, which handles emitting the cancelled
* event. And the stream is now in a bad state, so we can't reasonably
* expect to be able to send an error over it. */
});
this.stream.once('close', () => {
trace(
'Request to method ' +
this.handler?.path +
' stream closed with rstCode ' +
this.stream.rstCode
);
if (this.callEventTracker && !this.streamEnded) {
this.streamEnded = true;
this.callEventTracker.onStreamEnd(false);
this.callEventTracker.onCallEnd({
code: Status.CANCELLED,
details: 'Stream closed before sending status',
metadata: null
});
}
this.notifyOnCancel();
});
this.stream.on('data', (data: Buffer) => {
this.handleDataFrame(data);
});
this.stream.pause();
this.stream.on('end', () => {
this.handleEndEvent();
});
if ('grpc.max_send_message_length' in options) {
this.maxSendMessageSize = options['grpc.max_send_message_length']!;
}
if ('grpc.max_receive_message_length' in options) {
this.maxReceiveMessageSize = options['grpc.max_receive_message_length']!;
}
const metadata = Metadata.fromHttp2Headers(headers);
if (logging.isTracerEnabled(TRACER_NAME)) {
trace(
'Request to ' +
this.handler.path +
' received headers ' +
JSON.stringify(metadata.toJSON())
);
}
const timeoutHeader = metadata.get(GRPC_TIMEOUT_HEADER);
if (timeoutHeader.length > 0) {
this.handleTimeoutHeader(timeoutHeader[0] as string);
}
const encodingHeader = metadata.get(GRPC_ENCODING_HEADER);
if (encodingHeader.length > 0) {
this.incomingEncoding = encodingHeader[0] as string;
}
// Remove several headers that should not be propagated to the application
metadata.remove(GRPC_TIMEOUT_HEADER);
metadata.remove(GRPC_ENCODING_HEADER);
metadata.remove(GRPC_ACCEPT_ENCODING_HEADER);
metadata.remove(http2.constants.HTTP2_HEADER_ACCEPT_ENCODING);
metadata.remove(http2.constants.HTTP2_HEADER_TE);
metadata.remove(http2.constants.HTTP2_HEADER_CONTENT_TYPE);
this.metadata = metadata;
}
private handleTimeoutHeader(timeoutHeader: string) {
const match = timeoutHeader.toString().match(DEADLINE_REGEX);
if (match === null) {
const status: PartialStatusObject = {
code: Status.INTERNAL,
details: `Invalid ${GRPC_TIMEOUT_HEADER} value "${timeoutHeader}"`,
metadata: null
};
// Wait for the constructor to complete before sending the error.
process.nextTick(() => {
this.sendStatus(status);
});
return;
}
const timeout = (+match[1] * deadlineUnitsToMs[match[2]]) | 0;
const now = new Date();
this.deadline = now.setMilliseconds(now.getMilliseconds() + timeout);
this.deadlineTimer = setTimeout(() => {
const status: PartialStatusObject = {
code: Status.DEADLINE_EXCEEDED,
details: 'Deadline exceeded',
metadata: null
};
this.sendStatus(status);
}, timeout);
}
private checkCancelled(): boolean {
/* In some cases the stream can become destroyed before the close event
* fires. That creates a race condition that this check works around */
if (!this.cancelled && (this.stream.destroyed || this.stream.closed)) {
this.notifyOnCancel();
this.cancelled = true;
}
return this.cancelled;
}
private notifyOnCancel() {
if (this.cancelNotified) {
return;
}
this.cancelNotified = true;
this.cancelled = true;
process.nextTick(() => {
this.listener?.onCancel();
});
if (this.deadlineTimer) {
clearTimeout(this.deadlineTimer);
}
// Flush incoming data frames
this.stream.resume();
}
/**
* A server handler can start sending messages without explicitly sending
* metadata. In that case, we need to send headers before sending any
* messages. This function does that if necessary.
*/
private maybeSendMetadata() {
if (!this.metadataSent) {
this.sendMetadata(new Metadata());
}
}
/**
* Serialize a message to a length-delimited byte string.
* @param value
* @returns
*/
private serializeMessage(value: any) {
const messageBuffer = this.handler.serialize(value);
const byteLength = messageBuffer.byteLength;
const output = Buffer.allocUnsafe(byteLength + 5);
/* Note: response compression is currently not supported, so this
* compressed bit is always 0. */
output.writeUInt8(0, 0);
output.writeUInt32BE(byteLength, 1);
messageBuffer.copy(output, 5);
return output;
}
private decompressMessage(
message: Buffer,
encoding: string
): Buffer | Promise<Buffer> {
switch (encoding) {
case 'deflate':
return inflate(message.subarray(5));
case 'gzip':
return unzip(message.subarray(5));
case 'identity':
return message.subarray(5);
default:
return Promise.reject({
code: Status.UNIMPLEMENTED,
details: `Received message compressed with unsupported encoding "${encoding}"`,
});
}
}
private async decompressAndMaybePush(queueEntry: ReadQueueEntry) {
if (queueEntry.type !== 'COMPRESSED') {
throw new Error(`Invalid queue entry type: ${queueEntry.type}`);
}
const compressed = queueEntry.compressedMessage!.readUInt8(0) === 1;
const compressedMessageEncoding = compressed ? this.incomingEncoding : 'identity';
const decompressedMessage = await this.decompressMessage(queueEntry.compressedMessage!, compressedMessageEncoding);
try {
queueEntry.parsedMessage = this.handler.deserialize(decompressedMessage);
} catch (err) {
this.sendStatus({
code: Status.INTERNAL,
details: `Error deserializing request: ${(err as Error).message}`
});
return;
}
queueEntry.type = 'READABLE';
this.maybePushNextMessage();
}
private maybePushNextMessage() {
if (this.listener && this.isReadPending && this.readQueue.length > 0 && this.readQueue[0].type !== 'COMPRESSED') {
this.isReadPending = false;
const nextQueueEntry = this.readQueue.shift()!;
if (nextQueueEntry.type === 'READABLE') {
this.listener.onReceiveMessage(nextQueueEntry.parsedMessage);
} else {
// nextQueueEntry.type === 'HALF_CLOSE'
this.listener.onReceiveHalfClose();
}
}
}
private handleDataFrame(data: Buffer) {
if (this.checkCancelled()) {
return;
}
trace('Request to ' + this.handler.path + ' received data frame of size ' + data.length);
const rawMessages = this.decoder.write(data);
for (const messageBytes of rawMessages) {
this.stream.pause();
if (this.maxReceiveMessageSize !== -1 && messageBytes.length - 5 > this.maxReceiveMessageSize) {
this.sendStatus({
code: Status.RESOURCE_EXHAUSTED,
details: `Received message larger than max (${messageBytes.length - 5} vs. ${this.maxReceiveMessageSize})`,
metadata: null
});
return;
}
const queueEntry: ReadQueueEntry = {
type: 'COMPRESSED',
compressedMessage: messageBytes,
parsedMessage: null
};
this.readQueue.push(queueEntry);
this.decompressAndMaybePush(queueEntry);
this.callEventTracker?.addMessageReceived();
}
}
private handleEndEvent() {
this.readQueue.push({
type: 'HALF_CLOSE',
compressedMessage: null,
parsedMessage: null
});
this.receivedHalfClose = true;
this.maybePushNextMessage();
}
start(listener: InterceptingServerListener): void {
trace('Request to ' + this.handler.path + ' start called');
if (this.checkCancelled()) {
return;
}
this.listener = listener;
listener.onReceiveMetadata(this.metadata);
}
sendMetadata(metadata: Metadata): void {
if (this.checkCancelled()) {
return;
}
if (this.metadataSent) {
return;
}
this.metadataSent = true;
const custom = metadata ? metadata.toHttp2Headers() : null;
const headers = {
...defaultResponseHeaders,
...defaultCompressionHeaders,
...custom,
};
this.stream.respond(headers, defaultResponseOptions);
}
sendMessage(message: any, callback: () => void): void {
if (this.checkCancelled()) {
return;
}
let response: Buffer;
try {
response = this.serializeMessage(message);
} catch (e) {
this.sendStatus({
code: Status.INTERNAL,
details: `Error serializing response: ${getErrorMessage(e)}`,
metadata: null
});
return;
}
if (
this.maxSendMessageSize !== -1 &&
response.length - 5 > this.maxSendMessageSize
) {
this.sendStatus({
code: Status.RESOURCE_EXHAUSTED,
details: `Sent message larger than max (${response.length} vs. ${this.maxSendMessageSize})`,
metadata: null
});
return;
}
this.maybeSendMetadata();
trace('Request to ' + this.handler.path + ' sent data frame of size ' + response.length);
this.stream.write(response, error => {
if (error) {
this.sendStatus({
code: Status.INTERNAL,
details: `Error writing message: ${getErrorMessage(error)}`,
metadata: null
});
return;
}
this.callEventTracker?.addMessageSent();
callback();
});
}
sendStatus(status: PartialStatusObject): void {
if (this.checkCancelled()) {
return;
}
this.notifyOnCancel();
trace(
'Request to method ' +
this.handler?.path +
' ended with status code: ' +
Status[status.code] +
' details: ' +
status.details
);
if (this.stream.headersSent) {
if (!this.wantTrailers) {
this.wantTrailers = true;
this.stream.once('wantTrailers', () => {
if (this.callEventTracker && !this.streamEnded) {
this.streamEnded = true;
this.callEventTracker.onStreamEnd(true);
this.callEventTracker.onCallEnd(status);
}
const trailersToSend = {
[GRPC_STATUS_HEADER]: status.code,
[GRPC_MESSAGE_HEADER]: encodeURI(status.details),
...status.metadata?.toHttp2Headers(),
};
this.stream.sendTrailers(trailersToSend);
});
this.stream.end();
}
} else {
if (this.callEventTracker && !this.streamEnded) {
this.streamEnded = true;
this.callEventTracker.onStreamEnd(true);
this.callEventTracker.onCallEnd(status);
}
// Trailers-only response
const trailersToSend = {
[GRPC_STATUS_HEADER]: status.code,
[GRPC_MESSAGE_HEADER]: encodeURI(status.details),
...defaultResponseHeaders,
...status.metadata?.toHttp2Headers(),
};
this.stream.respond(trailersToSend, { endStream: true });
}
}
startRead(): void {
trace('Request to ' + this.handler.path + ' startRead called');
if (this.checkCancelled()) {
return;
}
this.isReadPending = true;
if (this.readQueue.length === 0) {
if (!this.receivedHalfClose) {
this.stream.resume();
}
} else {
this.maybePushNextMessage();
}
}
getPeer(): string {
const socket = this.stream.session?.socket;
if (socket?.remoteAddress) {
if (socket.remotePort) {
return `${socket.remoteAddress}:${socket.remotePort}`;
} else {
return socket.remoteAddress;
}
} else {
return 'unknown';
}
}
getDeadline(): Deadline {
return this.deadline;
}
}
export function getServerInterceptingCall(
interceptors: ServerInterceptor[],
stream: http2.ServerHttp2Stream,
headers: http2.IncomingHttpHeaders,
callEventTracker: CallEventTracker | null,
handler: Handler<any, any>,
options: ChannelOptions
) {
const methodDefinition: ServerMethodDefinition<any, any> = {
path: handler.path,
requestStream: handler.type === 'clientStream' || handler.type === 'bidi',
responseStream: handler.type === 'serverStream' || handler.type === 'bidi',
requestDeserialize: handler.deserialize,
responseSerialize: handler.serialize
}
const baseCall = new BaseServerInterceptingCall(stream, headers, callEventTracker, handler, options);
return interceptors.reduce((call: ServerInterceptingCallInterface, interceptor: ServerInterceptor) => {
return interceptor(methodDefinition, call);
}, baseCall);
}

420
packages/grpc-js/src/server.ts
View File

@ -28,20 +28,18 @@ import {
HandleCall, HandleCall,
Handler, Handler,
HandlerType, HandlerType,
Http2ServerCallStream,
sendUnaryData, sendUnaryData,
ServerDuplexStream, ServerDuplexStream,
ServerDuplexStreamImpl, ServerDuplexStreamImpl,
ServerReadableStream, ServerReadableStream,
ServerReadableStreamImpl,
ServerStreamingHandler, ServerStreamingHandler,
ServerUnaryCall, ServerUnaryCall,
ServerUnaryCallImpl,
ServerWritableStream, ServerWritableStream,
ServerWritableStreamImpl, ServerWritableStreamImpl,
UnaryHandler, UnaryHandler,
ServerErrorResponse, ServerErrorResponse,
ServerStatusResponse, ServerStatusResponse,
serverErrorToStatus,
} from './server-call'; } from './server-call';
import { ServerCredentials } from './server-credentials'; import { ServerCredentials } from './server-credentials';
import { ChannelOptions } from './channel-options'; import { ChannelOptions } from './channel-options';
@ -72,6 +70,9 @@ import {
unregisterChannelzRef, unregisterChannelzRef,
} from './channelz'; } from './channelz';
import { CipherNameAndProtocol, TLSSocket } from 'tls'; import { CipherNameAndProtocol, TLSSocket } from 'tls';
import { ServerInterceptingCallInterface, ServerInterceptor, getServerInterceptingCall } from './server-interceptors';
import { PartialStatusObject } from './call-interface';
import { CallEventTracker } from './transport';
const UNLIMITED_CONNECTION_AGE_MS = ~(1 << 31); const UNLIMITED_CONNECTION_AGE_MS = ~(1 << 31);
const KEEPALIVE_MAX_TIME_MS = ~(1 << 31); const KEEPALIVE_MAX_TIME_MS = ~(1 << 31);
@ -109,7 +110,7 @@ function deprecate(message: string) {
function getUnimplementedStatusResponse( function getUnimplementedStatusResponse(
methodName: string methodName: string
): Partial<ServiceError> { ): PartialStatusObject {
return { return {
code: Status.UNIMPLEMENTED, code: Status.UNIMPLEMENTED,
details: `The server does not implement the method ${methodName}`, details: `The server does not implement the method ${methodName}`,
@ -219,6 +220,10 @@ interface Http2ServerInfo {
sessions: Set<http2.ServerHttp2Session>; sessions: Set<http2.ServerHttp2Session>;
} }
export interface ServerOptions extends ChannelOptions {
interceptors?: ServerInterceptor[]
}
export class Server { export class Server {
private boundPorts: Map<string, BoundPort>= new Map(); private boundPorts: Map<string, BoundPort>= new Map();
private http2Servers: Map<AnyHttp2Server, Http2ServerInfo> = new Map(); private http2Servers: Map<AnyHttp2Server, Http2ServerInfo> = new Map();
@ -234,7 +239,7 @@ export class Server {
*/ */
private started = false; private started = false;
private shutdown = false; private shutdown = false;
private options: ChannelOptions; private options: ServerOptions;
private serverAddressString = 'null'; private serverAddressString = 'null';
// Channelz Info // Channelz Info
@ -251,13 +256,15 @@ export class Server {
private readonly keepaliveTimeMs: number; private readonly keepaliveTimeMs: number;
private readonly keepaliveTimeoutMs: number; private readonly keepaliveTimeoutMs: number;
private readonly interceptors: ServerInterceptor[];
/** /**
* Options that will be used to construct all Http2Server instances for this * Options that will be used to construct all Http2Server instances for this
* Server. * Server.
*/ */
private commonServerOptions: http2.ServerOptions; private commonServerOptions: http2.ServerOptions;
constructor(options?: ChannelOptions) { constructor(options?: ServerOptions) {
this.options = options ?? {}; this.options = options ?? {};
if (this.options['grpc.enable_channelz'] === 0) { if (this.options['grpc.enable_channelz'] === 0) {
this.channelzEnabled = false; this.channelzEnabled = false;
@ -296,6 +303,7 @@ export class Server {
maxConcurrentStreams: this.options['grpc.max_concurrent_streams'], maxConcurrentStreams: this.options['grpc.max_concurrent_streams'],
}; };
} }
this.interceptors = this.options.interceptors ?? [];
this.trace('Server constructed'); this.trace('Server constructed');
} }
@ -1072,23 +1080,24 @@ export class Server {
return handler; return handler;
} }
private _respondWithError<T extends Partial<ServiceError>>( private _respondWithError(
err: T, err: PartialStatusObject,
stream: http2.ServerHttp2Stream, stream: http2.ServerHttp2Stream,
channelzSessionInfo: ChannelzSessionInfo | null = null channelzSessionInfo: ChannelzSessionInfo | null = null
) { ) {
const call = new Http2ServerCallStream(stream, null!, this.options); const trailersToSend = {
'grpc-status': err.code ?? Status.INTERNAL,
if (err.code === undefined) { 'grpc-message': err.details,
err.code = Status.INTERNAL; [http2.constants.HTTP2_HEADER_STATUS]: http2.constants.HTTP_STATUS_OK,
} [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: 'application/grpc+proto',
...err.metadata?.toHttp2Headers()
};
stream.respond(trailersToSend, {endStream: true});
if (this.channelzEnabled) { if (this.channelzEnabled) {
this.callTracker.addCallFailed(); this.callTracker.addCallFailed();
channelzSessionInfo?.streamTracker.addCallFailed(); channelzSessionInfo?.streamTracker.addCallFailed();
} }
call.sendError(err);
} }
private _channelzHandler( private _channelzHandler(
@ -1120,39 +1129,44 @@ export class Server {
return; return;
} }
const call = new Http2ServerCallStream(stream, handler, this.options); let callEventTracker: CallEventTracker = {
addMessageSent: () => {
call.once('callEnd', (code: Status) => { if (channelzSessionInfo) {
if (code === Status.OK) { channelzSessionInfo.messagesSent += 1;
this.callTracker.addCallSucceeded(); channelzSessionInfo.lastMessageSentTimestamp = new Date();
} else {
this.callTracker.addCallFailed();
}
});
if (channelzSessionInfo) {
call.once('streamEnd', (success: boolean) => {
if (success) {
channelzSessionInfo.streamTracker.addCallSucceeded();
} else {
channelzSessionInfo.streamTracker.addCallFailed();
} }
}); },
call.on('sendMessage', () => { addMessageReceived: () => {
channelzSessionInfo.messagesSent += 1; if (channelzSessionInfo) {
channelzSessionInfo.lastMessageSentTimestamp = new Date(); channelzSessionInfo.messagesReceived += 1;
}); channelzSessionInfo.lastMessageReceivedTimestamp = new Date();
call.on('receiveMessage', () => { }
channelzSessionInfo.messagesReceived += 1; },
channelzSessionInfo.lastMessageReceivedTimestamp = new Date(); onCallEnd: status => {
}); if (status.code === Status.OK) {
this.callTracker.addCallSucceeded();
} else {
this.callTracker.addCallFailed();
}
},
onStreamEnd: success => {
if (channelzSessionInfo) {
if (success) {
channelzSessionInfo.streamTracker.addCallSucceeded();
} else {
channelzSessionInfo.streamTracker.addCallFailed();
}
}
}
} }
if (!this._runHandlerForCall(call, handler, headers)) { const call = getServerInterceptingCall(this.interceptors, stream, headers, callEventTracker, handler, this.options);
if (!this._runHandlerForCall(call, handler)) {
this.callTracker.addCallFailed(); this.callTracker.addCallFailed();
channelzSessionInfo?.streamTracker.addCallFailed(); channelzSessionInfo?.streamTracker.addCallFailed();
call.sendError({ call.sendStatus({
code: Status.INTERNAL, code: Status.INTERNAL,
details: `Unknown handler type: ${handler.type}`, details: `Unknown handler type: ${handler.type}`,
}); });
@ -1179,9 +1193,10 @@ export class Server {
return; return;
} }
const call = new Http2ServerCallStream(stream, handler, this.options); const call = getServerInterceptingCall(this.interceptors, stream, headers, null, handler, this.options);
if (!this._runHandlerForCall(call, handler, headers)) {
call.sendError({ if (!this._runHandlerForCall(call, handler)) {
call.sendStatus({
code: Status.INTERNAL, code: Status.INTERNAL,
details: `Unknown handler type: ${handler.type}`, details: `Unknown handler type: ${handler.type}`,
}); });
@ -1189,38 +1204,27 @@ export class Server {
} }
private _runHandlerForCall( private _runHandlerForCall(
call: Http2ServerCallStream<any, any>, call: ServerInterceptingCallInterface,
handler: Handler<any, any>, handler: Handler<any, any>
headers: http2.IncomingHttpHeaders
): boolean { ): boolean {
const metadata = call.receiveMetadata(headers);
const encoding =
(metadata.get('grpc-encoding')[0] as string | undefined) ?? 'identity';
metadata.remove('grpc-encoding');
const { type } = handler; const { type } = handler;
if (type === 'unary') { if (type === 'unary') {
handleUnary(call, handler as UntypedUnaryHandler, metadata, encoding); handleUnary(call, handler as UntypedUnaryHandler);
} else if (type === 'clientStream') { } else if (type === 'clientStream') {
handleClientStreaming( handleClientStreaming(
call, call,
handler as UntypedClientStreamingHandler, handler as UntypedClientStreamingHandler
metadata,
encoding
); );
} else if (type === 'serverStream') { } else if (type === 'serverStream') {
handleServerStreaming( handleServerStreaming(
call, call,
handler as UntypedServerStreamingHandler, handler as UntypedServerStreamingHandler
metadata,
encoding
); );
} else if (type === 'bidi') { } else if (type === 'bidi') {
handleBidiStreaming( handleBidiStreaming(
call, call,
handler as UntypedBidiStreamingHandler, handler as UntypedBidiStreamingHandler
metadata,
encoding
); );
} else { } else {
return false; return false;
@ -1365,52 +1369,10 @@ export class Server {
} }
async function handleUnary<RequestType, ResponseType>( async function handleUnary<RequestType, ResponseType>(
call: Http2ServerCallStream<RequestType, ResponseType>, call: ServerInterceptingCallInterface,
handler: UnaryHandler<RequestType, ResponseType>, handler: UnaryHandler<RequestType, ResponseType>
metadata: Metadata,
encoding: string
): Promise<void> { ): Promise<void> {
try { let stream: ServerUnaryCall<RequestType, ResponseType>;
const request = await call.receiveUnaryMessage(encoding);
if (request === undefined || call.cancelled) {
return;
}
const emitter = new ServerUnaryCallImpl<RequestType, ResponseType>(
call,
metadata,
request
);
handler.func(
emitter,
(
err: ServerErrorResponse | ServerStatusResponse | null,
value?: ResponseType | null,
trailer?: Metadata,
flags?: number
) => {
call.sendUnaryMessage(err, value, trailer, flags);
}
);
} catch (err) {
call.sendError(err as ServerErrorResponse);
}
}
function handleClientStreaming<RequestType, ResponseType>(
call: Http2ServerCallStream<RequestType, ResponseType>,
handler: ClientStreamingHandler<RequestType, ResponseType>,
metadata: Metadata,
encoding: string
): void {
const stream = new ServerReadableStreamImpl<RequestType, ResponseType>(
call,
metadata,
handler.deserialize,
encoding
);
function respond( function respond(
err: ServerErrorResponse | ServerStatusResponse | null, err: ServerErrorResponse | ServerStatusResponse | null,
@ -1418,61 +1380,207 @@ function handleClientStreaming<RequestType, ResponseType>(
trailer?: Metadata, trailer?: Metadata,
flags?: number flags?: number
) { ) {
stream.destroy(); if (err) {
call.sendUnaryMessage(err, value, trailer, flags); call.sendStatus(serverErrorToStatus(err, trailer));
}
if (call.cancelled) {
return;
}
stream.on('error', respond);
handler.func(stream, respond);
}
async function handleServerStreaming<RequestType, ResponseType>(
call: Http2ServerCallStream<RequestType, ResponseType>,
handler: ServerStreamingHandler<RequestType, ResponseType>,
metadata: Metadata,
encoding: string
): Promise<void> {
try {
const request = await call.receiveUnaryMessage(encoding);
if (request === undefined || call.cancelled) {
return; return;
} }
call.sendMessage(value, () => {
const stream = new ServerWritableStreamImpl<RequestType, ResponseType>( call.sendStatus({
call, code: Status.OK,
metadata, details: 'OK',
handler.serialize, metadata: trailer ?? null
request });
); });
handler.func(stream);
} catch (err) {
call.sendError(err as ServerErrorResponse);
} }
let requestMetadata: Metadata;
let requestMessage: RequestType | null = null;
call.start({
onReceiveMetadata(metadata) {
requestMetadata = metadata;
call.startRead();
},
onReceiveMessage(message) {
if (requestMessage) {
call.sendStatus({
code: Status.UNIMPLEMENTED,
details: `Received a second request message for server streaming method ${handler.path}`,
metadata: null
});
return;
}
requestMessage = message;
call.startRead();
},
onReceiveHalfClose() {
if (!requestMessage) {
call.sendStatus({
code: Status.UNIMPLEMENTED,
details: `Received no request message for server streaming method ${handler.path}`,
metadata: null
});
return;
}
stream = new ServerWritableStreamImpl(handler.path, call, requestMetadata, requestMessage);
try {
handler.func(stream, respond);
} catch (err) {
call.sendStatus({
code: Status.UNKNOWN,
details: `Server method handler threw error ${(err as Error).message}`,
metadata: null
});
}
},
onCancel() {
if (stream) {
stream.cancelled = true;
stream.emit('cancelled', 'cancelled');
}
},
});
}
function handleClientStreaming<RequestType, ResponseType>(
call: ServerInterceptingCallInterface,
handler: ClientStreamingHandler<RequestType, ResponseType>
): void {
let stream: ServerReadableStream<RequestType, ResponseType>;
function respond(
err: ServerErrorResponse | ServerStatusResponse | null,
value?: ResponseType | null,
trailer?: Metadata,
flags?: number
) {
if (err) {
call.sendStatus(serverErrorToStatus(err, trailer));
return;
}
call.sendMessage(value, () => {
call.sendStatus({
code: Status.OK,
details: 'OK',
metadata: trailer ?? null
});
});
}
call.start({
onReceiveMetadata(metadata) {
stream = new ServerDuplexStreamImpl(handler.path, call, metadata);
try {
handler.func(stream, respond);
} catch (err) {
call.sendStatus({
code: Status.UNKNOWN,
details: `Server method handler threw error ${(err as Error).message}`,
metadata: null
});
}
},
onReceiveMessage(message) {
stream.push(message);
},
onReceiveHalfClose() {
stream.push(null);
},
onCancel() {
if (stream) {
stream.cancelled = true;
stream.emit('cancelled', 'cancelled');
stream.destroy();
}
},
});
}
function handleServerStreaming<RequestType, ResponseType>(
call: ServerInterceptingCallInterface,
handler: ServerStreamingHandler<RequestType, ResponseType>
): void {
let stream: ServerWritableStream<RequestType, ResponseType>;
let requestMetadata: Metadata;
let requestMessage: RequestType | null = null;
call.start({
onReceiveMetadata(metadata) {
requestMetadata = metadata;
call.startRead();
},
onReceiveMessage(message) {
if (requestMessage) {
call.sendStatus({
code: Status.UNIMPLEMENTED,
details: `Received a second request message for server streaming method ${handler.path}`,
metadata: null
});
return;
}
requestMessage = message;
call.startRead();
},
onReceiveHalfClose() {
if (!requestMessage) {
call.sendStatus({
code: Status.UNIMPLEMENTED,
details: `Received no request message for server streaming method ${handler.path}`,
metadata: null
});
return;
}
stream = new ServerWritableStreamImpl(handler.path, call, requestMetadata, requestMessage);
try {
handler.func(stream);
} catch (err) {
call.sendStatus({
code: Status.UNKNOWN,
details: `Server method handler threw error ${(err as Error).message}`,
metadata: null
});
}
},
onCancel() {
if (stream) {
stream.cancelled = true;
stream.emit('cancelled', 'cancelled');
stream.destroy();
}
},
});
} }
function handleBidiStreaming<RequestType, ResponseType>( function handleBidiStreaming<RequestType, ResponseType>(
call: Http2ServerCallStream<RequestType, ResponseType>, call: ServerInterceptingCallInterface,
handler: BidiStreamingHandler<RequestType, ResponseType>, handler: BidiStreamingHandler<RequestType, ResponseType>
metadata: Metadata,
encoding: string
): void { ): void {
const stream = new ServerDuplexStreamImpl<RequestType, ResponseType>( let stream: ServerDuplexStream<RequestType, ResponseType>;
call,
metadata,
handler.serialize,
handler.deserialize,
encoding
);
if (call.cancelled) { call.start({
return; onReceiveMetadata(metadata) {
} stream = new ServerDuplexStreamImpl(handler.path, call, metadata);
try {
handler.func(stream); handler.func(stream);
} catch (err) {
call.sendStatus({
code: Status.UNKNOWN,
details: `Server method handler threw error ${(err as Error).message}`,
metadata: null
});
}
},
onReceiveMessage(message) {
stream.push(message);
},
onReceiveHalfClose() {
stream.push(null);
},
onCancel() {
if (stream) {
stream.cancelled = true;
stream.emit('cancelled', 'cancelled');
stream.destroy();
}
},
});
} }

4
packages/grpc-js/src/transport.ts
View File

@ -23,7 +23,7 @@ import {
PeerCertificate, PeerCertificate,
TLSSocket, TLSSocket,
} from 'tls'; } from 'tls';
import { StatusObject } from './call-interface'; import { PartialStatusObject } from './call-interface';
import { ChannelCredentials } from './channel-credentials'; import { ChannelCredentials } from './channel-credentials';
import { ChannelOptions } from './channel-options'; import { ChannelOptions } from './channel-options';
import { import {
@ -72,7 +72,7 @@ const KEEPALIVE_TIMEOUT_MS = 20000;
export interface CallEventTracker { export interface CallEventTracker {
addMessageSent(): void; addMessageSent(): void;
addMessageReceived(): void; addMessageReceived(): void;
onCallEnd(status: StatusObject): void; onCallEnd(status: PartialStatusObject): void;
onStreamEnd(success: boolean): void; onStreamEnd(success: boolean): void;
} }

7
packages/grpc-js/test/common.ts
View File

@ -72,7 +72,7 @@ const serviceImpl = {
export class TestServer { export class TestServer {
private server: grpc.Server; private server: grpc.Server;
public port: number | null = null; public port: number | null = null;
constructor(public useTls: boolean, options?: grpc.ChannelOptions) { constructor(public useTls: boolean, options?: grpc.ServerOptions) {
this.server = new grpc.Server(options); this.server = new grpc.Server(options);
this.server.addService(echoService.service, serviceImpl); this.server.addService(echoService.service, serviceImpl);
} }
@ -92,7 +92,6 @@ export class TestServer {
return; return;
} }
this.port = port; this.port = port;
this.server.start();
resolve(); resolve();
}); });
}); });
@ -130,6 +129,10 @@ export class TestClient {
this.client.echo({}, callback); this.client.echo({}, callback);
} }
sendRequestWithMetadata(metadata: grpc.Metadata, callback: (error?: grpc.ServiceError) => void) {
this.client.echo({}, metadata, callback);
}
getChannelState() { getChannelState() {
return this.client.getChannel().getConnectivityState(false); return this.client.getChannel().getConnectivityState(false);
} }

4
packages/grpc-js/test/test-channelz.ts
View File

@ -495,12 +495,12 @@ describe('Channelz', () => {
assert.strictEqual( assert.strictEqual(
+serverSocketResult.socket.data +serverSocketResult.socket.data
.streams_succeeded, .streams_succeeded,
0 1
); );
assert.strictEqual( assert.strictEqual(
+serverSocketResult.socket.data +serverSocketResult.socket.data
.streams_failed, .streams_failed,
1 0
); );
assert.strictEqual( assert.strictEqual(
+serverSocketResult.socket.data +serverSocketResult.socket.data

4
packages/grpc-js/test/test-server-deadlines.ts
View File

@ -110,8 +110,8 @@ describe('Server deadlines', () => {
{}, {},
(error: any, response: any) => { (error: any, response: any) => {
assert(error); assert(error);
assert.strictEqual(error.code, grpc.status.OUT_OF_RANGE); assert.strictEqual(error.code, grpc.status.INTERNAL);
assert.strictEqual(error.details, 'Invalid deadline'); assert.match(error.details, /^Invalid grpc-timeout value/);
done(); done();
} }
); );

296
packages/grpc-js/test/test-server-interceptors.ts Normal file
View File

@ -0,0 +1,296 @@
/*
* Copyright 2024 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
import * as assert from 'assert';
import * as path from 'path';
import * as grpc from '../src';
import { TestClient, loadProtoFile } from './common';
const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto');
const echoService = loadProtoFile(protoFile)
.EchoService as grpc.ServiceClientConstructor;
const AUTH_HEADER_KEY = 'auth';
const AUTH_HEADER_ALLOWED_VALUE = 'allowed';
const testAuthInterceptor: grpc.ServerInterceptor = (methodDescriptor, call) => {
return new grpc.ServerInterceptingCall(call, {
start: next => {
const authListener: grpc.ServerListener = {
onReceiveMetadata: (metadata, mdNext) => {
if (metadata.get(AUTH_HEADER_KEY)?.[0] !== AUTH_HEADER_ALLOWED_VALUE) {
call.sendStatus({
code: grpc.status.UNAUTHENTICATED,
details: 'Auth metadata not correct'
});
} else {
mdNext(metadata);
}
}
};
next(authListener);
}
});
};
let eventCounts = {
receiveMetadata: 0,
receiveMessage: 0,
receiveHalfClose: 0,
sendMetadata: 0,
sendMessage: 0,
sendStatus: 0
};
function resetEventCounts() {
eventCounts = {
receiveMetadata: 0,
receiveMessage: 0,
receiveHalfClose: 0,
sendMetadata: 0,
sendMessage: 0,
sendStatus: 0
};
}
/**
* Test interceptor to verify that interceptors see each expected event by
* counting each kind of event.
* @param methodDescription
* @param call
*/
const testLoggingInterceptor: grpc.ServerInterceptor = (methodDescription, call) => {
return new grpc.ServerInterceptingCall(call, {
start: next => {
next({
onReceiveMetadata: (metadata, mdNext) => {
eventCounts.receiveMetadata += 1;
mdNext(metadata);
},
onReceiveMessage: (message, messageNext) => {
eventCounts.receiveMessage += 1;
messageNext(message);
},
onReceiveHalfClose: hcNext => {
eventCounts.receiveHalfClose += 1;
hcNext();
}
});
},
sendMetadata: (metadata, mdNext) => {
eventCounts.sendMetadata += 1;
mdNext(metadata);
},
sendMessage: (message, messageNext) => {
eventCounts.sendMessage += 1;
messageNext(message);
},
sendStatus: (status, statusNext) => {
eventCounts.sendStatus += 1;
statusNext(status);
}
});
};
const testHeaderInjectionInterceptor: grpc.ServerInterceptor = (methodDescriptor, call) => {
return new grpc.ServerInterceptingCall(call, {
start: next => {
const authListener: grpc.ServerListener = {
onReceiveMetadata: (metadata, mdNext) => {
metadata.set('injected-header', 'present');
mdNext(metadata);
}
};
next(authListener);
}
});
};
describe('Server interceptors', () => {
describe('Auth-type interceptor', () => {
let server: grpc.Server;
let client: TestClient;
/* Tests that an interceptor can entirely prevent the handler from being
* invoked, based on the contents of the metadata. */
before(done => {
server = new grpc.Server({interceptors: [testAuthInterceptor]});
server.addService(echoService.service, {
echo: (
call: grpc.ServerUnaryCall<any, any>,
callback: grpc.sendUnaryData<any>
) => {
// A test will fail if a request makes it to the handler without the correct auth header
assert.strictEqual(call.metadata.get(AUTH_HEADER_KEY)?.[0], AUTH_HEADER_ALLOWED_VALUE);
callback(null, call.request);
},
});
server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
assert.ifError(error);
client = new TestClient(port, false);
done();
});
});
after(done => {
client.close();
server.tryShutdown(done);
});
it('Should accept a request with the expected header', done => {
const requestMetadata = new grpc.Metadata();
requestMetadata.set(AUTH_HEADER_KEY, AUTH_HEADER_ALLOWED_VALUE);
client.sendRequestWithMetadata(requestMetadata, done);
});
it('Should reject a request without the expected header', done => {
const requestMetadata = new grpc.Metadata();
requestMetadata.set(AUTH_HEADER_KEY, 'not allowed');
client.sendRequestWithMetadata(requestMetadata, error => {
assert.strictEqual(error?.code, grpc.status.UNAUTHENTICATED);
done();
});
});
});
describe('Logging-type interceptor', () => {
let server: grpc.Server;
let client: TestClient;
before(done => {
server = new grpc.Server({interceptors: [testLoggingInterceptor]});
server.addService(echoService.service, {
echo: (
call: grpc.ServerUnaryCall<any, any>,
callback: grpc.sendUnaryData<any>
) => {
call.sendMetadata(new grpc.Metadata());
callback(null, call.request);
},
});
server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
assert.ifError(error);
client = new TestClient(port, false);
done();
});
});
after(done => {
client.close();
server.tryShutdown(done);
});
beforeEach(() => {
resetEventCounts();
});
it('Should see every event once', done => {
client.sendRequest(error => {
assert.ifError(error);
assert.deepStrictEqual(eventCounts, {
receiveMetadata: 1,
receiveMessage: 1,
receiveHalfClose: 1,
sendMetadata: 1,
sendMessage: 1,
sendStatus: 1
});
done();
});
});
});
describe('Header injection interceptor', () => {
let server: grpc.Server;
let client: TestClient;
before(done => {
server = new grpc.Server({interceptors: [testHeaderInjectionInterceptor]});
server.addService(echoService.service, {
echo: (
call: grpc.ServerUnaryCall<any, any>,
callback: grpc.sendUnaryData<any>
) => {
assert.strictEqual(call.metadata.get('injected-header')?.[0], 'present');
callback(null, call.request);
},
});
server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
assert.ifError(error);
client = new TestClient(port, false);
done();
});
});
after(done => {
client.close();
server.tryShutdown(done);
});
it('Should inject the header for the handler to see', done => {
client.sendRequest(done);
});
});
describe('Multiple interceptors', () => {
let server: grpc.Server;
let client: TestClient;
before(done => {
server = new grpc.Server({interceptors: [testAuthInterceptor, testLoggingInterceptor, testHeaderInjectionInterceptor]});
server.addService(echoService.service, {
echo: (
call: grpc.ServerUnaryCall<any, any>,
callback: grpc.sendUnaryData<any>
) => {
assert.strictEqual(call.metadata.get(AUTH_HEADER_KEY)?.[0], AUTH_HEADER_ALLOWED_VALUE);
assert.strictEqual(call.metadata.get('injected-header')?.[0], 'present');
call.sendMetadata(new grpc.Metadata());
callback(null, call.request);
},
});
server.bindAsync('localhost:0', grpc.ServerCredentials.createInsecure(), (error, port) => {
assert.ifError(error);
client = new TestClient(port, false);
done();
});
});
after(done => {
client.close();
server.tryShutdown(done);
});
beforeEach(() => {
resetEventCounts();
});
it('Should not log requests rejected by auth', done => {
const requestMetadata = new grpc.Metadata();
requestMetadata.set(AUTH_HEADER_KEY, 'not allowed');
client.sendRequestWithMetadata(requestMetadata, error => {
assert.strictEqual(error?.code, grpc.status.UNAUTHENTICATED);
assert.deepStrictEqual(eventCounts, {
receiveMetadata: 0,
receiveMessage: 0,
receiveHalfClose: 0,
sendMetadata: 0,
sendMessage: 0,
sendStatus: 0
});
done();
});
});
it('Should log requests accepted by auth', done => {
const requestMetadata = new grpc.Metadata();
requestMetadata.set(AUTH_HEADER_KEY, AUTH_HEADER_ALLOWED_VALUE);
client.sendRequestWithMetadata(requestMetadata, error => {
assert.ifError(error);
assert.deepStrictEqual(eventCounts, {
receiveMetadata: 1,
receiveMessage: 1,
receiveHalfClose: 1,
sendMetadata: 1,
sendMessage: 1,
sendStatus: 1
});
done();
});
});
});
});