Merge pull request #370 from MatrixAI/GRPC_call_failure_fix

fix: bug with GPRC client call through proxy

Author: CMCDragonkai

src/network/ConnectionReverse.ts

@@ -295,7 +295,9 @@ class ConnectionReverse extends Connection {
         await this.stop();
       });
       this.tlsSocket.pipe(this.serverSocket, { end: false });
-      this.serverSocket.pipe(this.tlsSocket, { end: false });
+      this.tlsSocket.once('data', () => {
+        this.serverSocket.pipe(this.tlsSocket as TLSSocket, { end: false });
+      });
       this.clientCertChain = clientCertChain;
       this.logger.info('Composed Connection Reverse');
     } catch (e) {

src/network/Proxy.ts

@@ -434,6 +434,13 @@ class Proxy {
           timer,
         );
       } catch (e) {
+        if (e instanceof networkErrors.ErrorProxyConnectInvalidUrl) {
+          if (!clientSocket.destroyed) {
+            await clientSocketEnd('HTTP/1.1 400 Bad Request\r\n' + '\r\n');
+            clientSocket.destroy(e);
+          }
+          return;
+        }
         if (e instanceof networkErrors.ErrorConnectionStartTimeout) {
           if (!clientSocket.destroyed) {
             await clientSocketEnd('HTTP/1.1 504 Gateway Timeout\r\n' + '\r\n');
@@ -519,6 +526,9 @@ class Proxy {
     proxyPort: Port,
     timer?: Timer,
   ): Promise<ConnectionForward> {
+    if (networkUtils.isHostWildcard(proxyHost)) {
+      throw new networkErrors.ErrorProxyConnectInvalidUrl();
+    }
     const proxyAddress = networkUtils.buildAddress(proxyHost, proxyPort);
     let conn: ConnectionForward | undefined;
     conn = this.connectionsForward.proxy.get(proxyAddress);
@@ -681,6 +691,9 @@ class Proxy {
     proxyPort: Port,
     timer?: Timer,
   ): Promise<ConnectionReverse> {
+    if (networkUtils.isHostWildcard(proxyHost)) {
+      throw new networkErrors.ErrorProxyConnectInvalidUrl();
+    }
     const proxyAddress = networkUtils.buildAddress(proxyHost, proxyPort);
     let conn = this.connectionsReverse.proxy.get(proxyAddress);
     if (conn != null) {

src/network/utils.ts

@@ -29,6 +29,10 @@ function isHost(host: any): host is Host {
   return isIPv4 || isIPv6;
 }
 
+function isHostWildcard(host: Host): boolean {
+  return host === '0.0.0.0' || host === '::';
+}
+
 /**
  * Validates hostname as per RFC 1123
  */
@@ -353,6 +357,7 @@ export {
   pingBuffer,
   pongBuffer,
   isHost,
+  isHostWildcard,
   isHostname,
   isPort,
   toAuthToken,

src/nodes/NodeConnection.ts

@@ -59,6 +59,10 @@ class NodeConnection<T extends GRPCClient> {
     logger?: Logger;
   }): Promise<NodeConnection<T>> {
     logger.info(`Creating ${this.name}`);
+    // Checking if attempting to connect to a wildcard IP
+    if (networkUtils.isHostWildcard(targetHost)) {
+      throw new nodesErrors.ErrorNodeConnectionHostWildcard();
+    }
     const proxyConfig = {
       host: proxy.getForwardHost(),
       port: proxy.getForwardPort(),

src/nodes/NodeConnectionManager.ts

@@ -129,7 +129,7 @@ class NodeConnectionManager {
     targetNodeId: NodeId,
   ): Promise<ResourceAcquire<NodeConnection<GRPCClientAgent>>> {
     return async () => {
-      const connAndLock = await this.createConnection(targetNodeId);
+      const connAndLock = await this.getConnection(targetNodeId);
       // Acquire the read lock and the release function
       const release = await connAndLock.lock.acquireRead();
       // Resetting TTL timer
@@ -161,6 +161,11 @@ class NodeConnectionManager {
       return await withF(
         [await this.acquireConnection(targetNodeId)],
         async ([conn]) => {
+          this.logger.info(
+            `withConnF calling function with connection to ${nodesUtils.encodeNodeId(
+              targetNodeId,
+            )}`,
+          );
           return await f(conn);
         },
       );
@@ -219,11 +224,11 @@ class NodeConnectionManager {
    * @param targetNodeId Id of node we are creating connection to
    * @returns ConnectionAndLock that was create or exists in the connection map.
    */
-  protected async createConnection(
+  protected async getConnection(
     targetNodeId: NodeId,
   ): Promise<ConnectionAndLock> {
     this.logger.info(
-      `Creating connection to ${nodesUtils.encodeNodeId(targetNodeId)}`,
+      `Getting connection to ${nodesUtils.encodeNodeId(targetNodeId)}`,
     );
     let connection: NodeConnection<GRPCClientAgent> | undefined;
     let lock: RWLock;
@@ -241,10 +246,13 @@ class NodeConnectionManager {
           targetNodeId.toString() as NodeIdString,
         );
         if (connAndLock != null && connAndLock.connection != null) {
+          this.logger.info(
+            `existing entry found for ${nodesUtils.encodeNodeId(targetNodeId)}`,
+          );
           return connAndLock;
         }
         this.logger.info(
-          `existing lock: creating connection to ${nodesUtils.encodeNodeId(
+          `existing lock, creating connection to ${nodesUtils.encodeNodeId(
             targetNodeId,
           )}`,
         );
@@ -260,7 +268,7 @@ class NodeConnectionManager {
       );
       return await lock.withWrite(async () => {
         this.logger.info(
-          `no existing entry: creating connection to ${nodesUtils.encodeNodeId(
+          `no existing entry, creating connection to ${nodesUtils.encodeNodeId(
             targetNodeId,
           )}`,
         );
@@ -318,7 +326,9 @@ class NodeConnectionManager {
       nodeConnectionManager: this,
       destroyCallback,
       connConnectTime: this.connConnectTime,
-      logger: this.logger.getChild(`${targetHost}:${targetAddress.port}`),
+      logger: this.logger.getChild(
+        `${NodeConnection.name} ${targetHost}:${targetAddress.port}`,
+      ),
       clientFactory: async (args) =>
         GRPCClientAgent.createGRPCClientAgent(args),
     });
@@ -521,7 +531,7 @@ class NodeConnectionManager {
         // Add the node to the database so that we can find its address in
         // call to getConnectionToNode
         await this.nodeGraph.setNode(nextNode.id, nextNode.address);
-        await this.createConnection(nextNode.id);
+        await this.getConnection(nextNode.id);
       } catch (e) {
         // If we can't connect to the node, then skip it
         continue;
@@ -620,7 +630,7 @@ class NodeConnectionManager {
     for (const seedNodeId of this.getSeedNodes()) {
       // Check if the connection is viable
       try {
-        await this.createConnection(seedNodeId);
+        await this.getConnection(seedNodeId);
       } catch (e) {
         if (e instanceof nodesErrors.ErrorNodeConnectionTimeout) continue;
         throw e;

src/nodes/errors.ts

@@ -62,6 +62,11 @@ class ErrorNodeConnectionManagerNotRunning extends ErrorNodes {
   exitCode = sysexits.USAGE;
 }
 
+class ErrorNodeConnectionHostWildcard extends ErrorNodes {
+  description = 'An IP wildcard was provided for the target host';
+  exitCode = sysexits.USAGE;
+}
+
 export {
   ErrorNodes,
   ErrorNodeGraphRunning,
@@ -76,4 +81,5 @@ export {
   ErrorNodeConnectionInfoNotExist,
   ErrorNodeConnectionPublicKeyNotFound,
   ErrorNodeConnectionManagerNotRunning,
+  ErrorNodeConnectionHostWildcard,
 };

tests/agent/service/nodesCrossSignClaim.test.ts

@@ -52,6 +52,9 @@ describe('nodesCrossSignClaim', () => {
         rootKeyPairBits: 2048,
       },
       seedNodes: {}, // Explicitly no seed nodes on startup
+      networkConfig: {
+        proxyHost: '127.0.0.1' as Host,
+      },
       logger,
     });
     localId = pkAgent.keyManager.getNodeId();
@@ -63,6 +66,9 @@ describe('nodesCrossSignClaim', () => {
         rootKeyPairBits: 2048,
       },
       seedNodes: {}, // Explicitly no seed nodes on startup
+      networkConfig: {
+        proxyHost: '127.0.0.1' as Host,
+      },
       logger,
     });
     remoteId = remoteNode.keyManager.getNodeId();

tests/bin/vaults/vaults.test.ts

@@ -1,5 +1,6 @@
 import type { NodeIdEncoded, NodeAddress, NodeInfo } from '@/nodes/types';
 import type { VaultId, VaultName } from '@/vaults/types';
+import type { Host } from '@/network/types';
 import os from 'os';
 import path from 'path';
 import fs from 'fs';
@@ -207,6 +208,9 @@ describe('CLI vaults', () => {
       const targetPolykeyAgent = await PolykeyAgent.createPolykeyAgent({
         password,
         nodePath: dataDir2,
+        networkConfig: {
+          proxyHost: '127.0.0.1' as Host,
+        },
         logger: logger,
       });
       const vaultId = await targetPolykeyAgent.vaultManager.createVault(
@@ -701,6 +705,9 @@ describe('CLI vaults', () => {
             password,
             logger,
             nodePath: path.join(dataDir, 'remoteOnline'),
+            networkConfig: {
+              proxyHost: '127.0.0.1' as Host,
+            },
           });
           const remoteOnlineNodeId = remoteOnline.keyManager.getNodeId();
           const remoteOnlineNodeIdEncoded =

tests/network/Proxy.test.ts

@@ -208,6 +208,15 @@ describe(Proxy.name, () => {
         `127.0.0.1:80?nodeId=${encodeURIComponent(nodeIdSomeEncoded)}`,
       ),
     ).rejects.toThrow('407');
+    // Wildcard as host
+    await expect(() =>
+      httpConnect(
+        proxy.getForwardHost(),
+        proxy.getForwardPort(),
+        authToken,
+        `0.0.0.0:80?nodeId=${encodeURIComponent(nodeIdSomeEncoded)}`,
+      ),
+    ).rejects.toThrow('400');
     // No node id
     await expect(() =>
       httpConnect(

tests/nodes/NodeConnection.test.ts

@@ -433,6 +433,47 @@ describe(`${NodeConnection.name} test`, () => {
     });
     await conn.destroy();
   });
+  test('connects to its target but proxies connect first', async () => {
+    await clientproxy.openConnectionForward(
+      targetNodeId,
+      localHost,
+      targetPort,
+    );
+    const conn = await NodeConnection.createNodeConnection({
+      targetNodeId: targetNodeId,
+      targetHost: localHost,
+      targetPort: targetPort,
+      proxy: clientproxy,
+      keyManager: clientKeyManager,
+      nodeConnectionManager: dummyNodeConnectionManager,
+      destroyCallback,
+      logger: logger,
+      clientFactory: async (args) =>
+        GRPCClientAgent.createGRPCClientAgent(args),
+    });
+    // Because the connection will not have enough time to compose before we
+    // attempt to acquire the connection info, we need to wait and poll it
+    const connInfo = await poll<ConnectionInfo | undefined>(
+      async () => {
+        return serverProxy.getConnectionInfoByProxy(localHost, sourcePort);
+      },
+      (e) => {
+        if (e instanceof networkErrors.ErrorConnectionNotComposed) return false;
+        if (e instanceof networkErrors.ErrorConnectionNotRunning) return false;
+        return true;
+      },
+    );
+    expect(connInfo).toBeDefined();
+    expect(connInfo).toMatchObject({
+      remoteNodeId: sourceNodeId,
+      remoteCertificates: expect.any(Array),
+      localHost: localHost,
+      localPort: targetPort,
+      remoteHost: localHost,
+      remotePort: sourcePort,
+    });
+    await conn.destroy();
+  });
   test('grpcCall after connection drops', async () => {
     let nodeConnection: NodeConnection<GRPCClientAgent> | undefined;
     let polykeyAgent: PolykeyAgent | undefined;
@@ -441,6 +482,9 @@ describe(`${NodeConnection.name} test`, () => {
         password,
         nodePath: path.join(dataDir, 'PolykeyAgent3'),
         logger: logger,
+        networkConfig: {
+          proxyHost: localHost,
+        },
       });
       // Have a nodeConnection try to connect to it
       const killSelf = jest.fn();
@@ -629,6 +673,9 @@ describe(`${NodeConnection.name} test`, () => {
         password,
         nodePath: path.join(dataDir, 'PolykeyAgent3'),
         logger: logger,
+        networkConfig: {
+          proxyHost: localHost,
+        },
       });
       // Have a nodeConnection try to connect to it
       const killSelf = jest.fn();