grpc-js: Expand ServerCredentials API to support watchers

Author: murgatroid99

packages/grpc-js/src/server-credentials.ts

@@ -17,15 +17,40 @@
 
 import { SecureServerOptions } from 'http2';
 import { CIPHER_SUITES, getDefaultRootsData } from './tls-helpers';
+import { SecureContextOptions } from 'tls';
 
 export interface KeyCertPair {
   private_key: Buffer;
   cert_chain: Buffer;
 }
 
+export interface SecureContextWatcher {
+  (context: SecureContextOptions | null): void;
+}
+
 export abstract class ServerCredentials {
+  private watchers: Set<SecureContextWatcher> = new Set();
+  private latestContextOptions: SecureServerOptions | null = null;
+  _addWatcher(watcher: SecureContextWatcher) {
+    this.watchers.add(watcher);
+  }
+  _removeWatcher(watcher: SecureContextWatcher) {
+    this.watchers.delete(watcher);
+  }
+  protected updateSecureContextOptions(options: SecureServerOptions | null) {
+    if (options) {
+      this.latestContextOptions = options;
+    } else {
+      this.latestContextOptions = null;
+    }
+    for (const watcher of this.watchers) {
+      watcher(this.latestContextOptions);
+    }
+  }
   abstract _isSecure(): boolean;
-  abstract _getSettings(): SecureServerOptions | null;
+  _getSettings(): SecureServerOptions | null {
+    return this.latestContextOptions;
+  }
   abstract _equals(other: ServerCredentials): boolean;
 
   static createInsecure(): ServerCredentials {

packages/grpc-js/src/server.ts

@@ -41,7 +41,7 @@ import {
   ServerStatusResponse,
   serverErrorToStatus,
 } from './server-call';
-import { ServerCredentials } from './server-credentials';
+import { SecureContextWatcher, ServerCredentials } from './server-credentials';
 import { ChannelOptions } from './channel-options';
 import {
   createResolver,
@@ -73,6 +73,7 @@ import { CipherNameAndProtocol, TLSSocket } from 'tls';
 import { ServerInterceptingCallInterface, ServerInterceptor, getServerInterceptingCall } from './server-interceptors';
 import { PartialStatusObject } from './call-interface';
 import { CallEventTracker } from './transport';
+import { Socket } from 'net';
 
 const UNLIMITED_CONNECTION_AGE_MS = ~(1 << 31);
 const KEEPALIVE_MAX_TIME_MS = ~(1 << 31);
@@ -501,13 +502,19 @@ export class Server {
   private createHttp2Server(credentials: ServerCredentials) {
     let http2Server: http2.Http2Server | http2.Http2SecureServer;
     if (credentials._isSecure()) {
-      const secureServerOptions = Object.assign(
-        this.commonServerOptions,
-        credentials._getSettings()!
-      );
-      secureServerOptions.enableTrace =
-        this.options['grpc-node.tls_enable_trace'] === 1;
+      const credentialsSettings = credentials._getSettings();
+      const secureServerOptions: http2.SecureServerOptions = {
+        ...this.commonServerOptions,
+        ...credentialsSettings,
+        enableTrace: this.options['grpc-node.tls_enable_trace'] === 1
+      };
+      let areCredentialsValid = credentialsSettings !== null;
       http2Server = http2.createSecureServer(secureServerOptions);
+      http2Server.on('connection', (socket: Socket) => {
+        if (!areCredentialsValid) {
+          socket.destroy();
+        }
+      });
       http2Server.on('secureConnection', (socket: TLSSocket) => {
         /* These errors need to be handled by the user of Http2SecureServer,
          * according to https://github.com/nodejs/node/issues/35824 */
@@ -517,6 +524,16 @@ export class Server {
           );
         });
       });
+      const credsWatcher: SecureContextWatcher = options => {
+        if (options) {
+          (http2Server as http2.Http2SecureServer).setSecureContext(options);
+        }
+        areCredentialsValid = options !== null;
+      }
+      credentials._addWatcher(credsWatcher);
+      http2Server.on('close', () => {
+        credentials._removeWatcher(credsWatcher);
+      });
     } else {
       http2Server = http2.createServer(this.commonServerOptions);
     }

packages/grpc-js/test/test-server.ts

@@ -41,6 +41,7 @@ import {
 import { ProtoGrpcType as TestServiceGrpcType } from './generated/test_service';
 import { Request__Output } from './generated/Request';
 import { CompressionAlgorithms } from '../src/compression-algorithms';
+import { SecureContextOptions } from 'tls';
 
 const loadedTestServiceProto = protoLoader.loadSync(
   path.join(__dirname, 'fixtures/test_service.proto'),
@@ -746,6 +747,116 @@ describe('Echo service', () => {
     );
   });
 
+  describe('ServerCredentials watcher', () => {
+    let server: Server;
+    let serverPort: number;
+    const protoFile = path.join(__dirname, 'fixtures', 'echo_service.proto');
+    const echoService = loadProtoFile(protoFile)
+      .EchoService as ServiceClientConstructor;
+
+    class ToggleableSecureServerCredentials extends ServerCredentials {
+      private contextOptions: SecureContextOptions;
+      constructor(key: Buffer, cert: Buffer) {
+        super();
+        this.contextOptions = {key, cert};
+        this.enable();
+      }
+      enable() {
+        this.updateSecureContextOptions(this.contextOptions);
+      }
+      disable() {
+        this.updateSecureContextOptions(null);
+      }
+      _isSecure(): boolean {
+        return true;
+      }
+      _equals(other: grpc.ServerCredentials): boolean {
+        return this === other;
+      }
+    }
+
+    const serverCredentials = new ToggleableSecureServerCredentials(key, cert);
+
+    const serviceImplementation = {
+      echo(call: ServerUnaryCall<any, any>, callback: sendUnaryData<any>) {
+        callback(null, call.request);
+      },
+      echoBidiStream(call: ServerDuplexStream<any, any>) {
+        call.on('data', data => {
+          call.write(data);
+        });
+        call.on('end', () => {
+          call.end();
+        });
+      },
+    };
+
+    before(done => {
+      server = new Server();
+      server.addService(echoService.service, serviceImplementation);
+
+      server.bindAsync(
+        'localhost:0',
+        serverCredentials,
+        (err, port) => {
+          assert.ifError(err);
+          serverPort = port;
+          done();
+        }
+      );
+    });
+
+    after(done => {
+      client.close();
+      server.tryShutdown(done);
+    });
+
+    it('should make successful requests only when the credentials are enabled', done => {
+      const client1 = new echoService(
+        `localhost:${serverPort}`,
+        grpc.credentials.createSsl(ca),
+        {
+          'grpc.ssl_target_name_override': 'foo.test.google.fr',
+          'grpc.default_authority': 'foo.test.google.fr',
+          'grpc.use_local_subchannel_pool': 1
+        }
+      );
+      const testMessage = { value: 'test value', value2: 3 };
+      client1.echo(testMessage, (error: ServiceError, response: any) => {
+        assert.ifError(error);
+        assert.deepStrictEqual(response, testMessage);
+        serverCredentials.disable();
+        const client2 = new echoService(
+          `localhost:${serverPort}`,
+          grpc.credentials.createSsl(ca),
+          {
+            'grpc.ssl_target_name_override': 'foo.test.google.fr',
+            'grpc.default_authority': 'foo.test.google.fr',
+            'grpc.use_local_subchannel_pool': 1
+          }
+        );
+        client2.echo(testMessage, (error: ServiceError, response: any) => {
+          assert(error);
+          assert.strictEqual(error.code, grpc.status.UNAVAILABLE);
+          serverCredentials.enable();
+          const client3 = new echoService(
+            `localhost:${serverPort}`,
+            grpc.credentials.createSsl(ca),
+            {
+              'grpc.ssl_target_name_override': 'foo.test.google.fr',
+              'grpc.default_authority': 'foo.test.google.fr',
+              'grpc.use_local_subchannel_pool': 1
+            }
+          );
+          client3.echo(testMessage, (error: ServiceError, response: any) => {
+            assert.ifError(error);
+            done();
+          });
+        });
+      });
+    });
+  });
+
   /* This test passes on Node 18 but fails on Node 16. The failure appears to
    * be caused by https://github.com/nodejs/node/issues/42713 */
   it.skip('should continue a stream after server shutdown', done => {