feat: added duplex stream for connection authentication

chore: updated documentation

fix: lint

fix: import order

fix: all tests now passing

fix: auth handler hanging until timeout if errored

Author: aryanjassal

src/nodes/NodeConnectionManager.ts

@@ -14,12 +14,7 @@ import type {
 } from '../tasks/types.js';
 import type { SignedTokenEncoded } from '../tokens/types.js';
 import type { Host, Port } from '../network/types.js';
-import type {
-  Claim,
-  ClaimId,
-  // ClaimIdEncoded,
-  SignedClaim,
-} from '../claims/types.js';
+import type { Claim, ClaimId, SignedClaim } from '../claims/types.js';
 import type { ClaimLinkNode } from '../claims/payloads/index.js';
 import type NodeConnection from '../nodes/NodeConnection.js';
 import type {
@@ -48,8 +43,8 @@ import { decorators } from '@matrixai/contexts';
 import * as nodesUtils from './utils.js';
 import * as nodesEvents from './events.js';
 import * as nodesErrors from './errors.js';
-import * as agentErrors from './agent/errors.js';
 import NodeConnectionQueue from './NodeConnectionQueue.js';
+import config from '../config.js';
 import { assertClaimNetworkAuthority } from '../claims/payloads/claimNetworkAuthority.js';
 import { assertClaimNetworkAccess } from '../claims/payloads/claimNetworkAccess.js';
 import Token from '../tokens/Token.js';
@@ -58,7 +53,6 @@ import * as tasksErrors from '../tasks/errors.js';
 import * as claimsUtils from '../claims/utils.js';
 import * as claimsErrors from '../claims/errors.js';
 import * as utils from '../utils/utils.js';
-import config from '../config.js';
 import * as networkUtils from '../network/utils.js';
 
 const abortEphemeralTaskReason = Symbol('abort ephemeral task reason');
@@ -1628,7 +1622,7 @@ class NodeManager {
     }
 
     if (!success) {
-      throw new agentErrors.ErrorNodesClaimNetworkVerificationFailed();
+      throw new nodesErrors.ErrorNodeClaimNetworkVerificationFailed();
     }
 
     return {

src/nodes/NodeManager.ts

@@ -8,28 +8,4 @@ class ErrorAgentNodeIdMissing<T> extends ErrorAgent<T> {
   exitCode = sysexits.UNAVAILABLE;
 }
 
-class ErrorNodesConnectionSignalRequestVerificationFailed<
-  T,
-> extends ErrorAgent<T> {
-  static description = 'Failed to verify request message signature';
-  exitCode = sysexits.UNAVAILABLE;
-}
-
-class ErrorNodesConnectionSignalRelayVerificationFailed<
-  T,
-> extends ErrorAgent<T> {
-  static description = 'Failed to verify relay message signature';
-  exitCode = sysexits.UNAVAILABLE;
-}
-
-class ErrorNodesClaimNetworkVerificationFailed<T> extends ErrorAgent<T> {
-  static description = 'Failed to verify claim network message';
-  exitCode = sysexits.UNAVAILABLE;
-}
-
-export {
-  ErrorAgentNodeIdMissing,
-  ErrorNodesConnectionSignalRequestVerificationFailed,
-  ErrorNodesConnectionSignalRelayVerificationFailed,
-  ErrorNodesClaimNetworkVerificationFailed,
-};
+export { ErrorAgentNodeIdMissing };

src/nodes/agent/errors.ts

@@ -42,42 +42,13 @@ class NodesAuthenticateConnection extends DuplexHandler<
       throw new agentErrors.ErrorAgentNodeIdMissing();
     }
 
-    // Forward authentication message processing
-    const {
-      value: forwardMessageIn,
-    }: {
-      value: NodesAuthenticateConnectionMessage | SuccessMessage;
-    } = await input.next();
-    if (forwardMessageIn.type === 'success') throw new Error('exit');
-    const forwardMessageOut = await nodeConnectionManager.handleAuthentication(
+    // This async generator handles the back-and-forth communication to
+    // authenticate a connection.
+    yield* nodeConnectionManager.handleAuthentication(
       requestingNodeId,
-      forwardMessageIn,
+      input,
       ctx,
     );
-    yield {
-      type: 'success',
-      success: true,
-    };
-
-    // Sending authentication message
-    yield forwardMessageOut;
-    const {
-      value: reverseMessageIn,
-    }: {
-      value: NodesAuthenticateConnectionMessage | SuccessMessage;
-    } = await input.next();
-    if (reverseMessageIn.type !== 'success') throw new Error('exit');
-
-    nodeConnectionManager.finalizeAuthentication(
-      requestingNodeId,
-      reverseMessageIn.success,
-    );
-    yield {
-      type: 'success',
-      success: true,
-    }
-    // success: true
-    // fire authentication events before final ack
   };
 }
 

src/nodes/agent/handlers/NodesAuthenticateConnection.ts

@@ -15,6 +15,7 @@ import * as keysUtils from '../../../keys/utils/index.js';
 import * as ids from '../../../ids/index.js';
 import * as agentErrors from '../errors.js';
 import * as agentUtils from '../utils.js';
+import * as nodesErrors from '../../errors.js';
 
 class NodesConnectionSignalFinal extends UnaryHandler<
   {
@@ -68,7 +69,7 @@ class NodesConnectionSignalFinal extends UnaryHandler<
         requestSignature,
       )
     ) {
-      throw new agentErrors.ErrorNodesConnectionSignalRequestVerificationFailed();
+      throw new nodesErrors.ErrorNodeConnectionSignalRequestVerificationFailed();
     }
     // Checking relay message relaySignature.
     // relayData is just `<sourceNodeId><targetNodeId><Address><requestSignature>` concatenated.
@@ -83,7 +84,7 @@ class NodesConnectionSignalFinal extends UnaryHandler<
     if (
       !keysUtils.verifyWithPublicKey(relayPublicKey, relayData, relaySignature)
     ) {
-      throw new agentErrors.ErrorNodesConnectionSignalRelayVerificationFailed();
+      throw new nodesErrors.ErrorNodeConnectionSignalRelayVerificationFailed();
     }
 
     const host = input.address.host as Host;

src/nodes/agent/handlers/NodesConnectionSignalFinal.ts

@@ -14,6 +14,7 @@ import { matchSync } from '../../../utils/index.js';
 import { never } from '../../../utils/index.js';
 import * as agentErrors from '../errors.js';
 import * as agentUtils from '../utils.js';
+import * as nodesErrors from '../../errors.js';
 import * as keysUtils from '../../../keys/utils/index.js';
 import * as ids from '../../../ids/index.js';
 
@@ -55,7 +56,7 @@ class NodesConnectionSignalInitial extends UnaryHandler<
     const data = Buffer.concat([requestingNodeId, targetNodeId]);
     const sourcePublicKey = keysUtils.publicKeyFromNodeId(requestingNodeId);
     if (!keysUtils.verifyWithPublicKey(sourcePublicKey, data, signature)) {
-      throw new agentErrors.ErrorNodesConnectionSignalRelayVerificationFailed();
+      throw new nodesErrors.ErrorNodeConnectionSignalRelayVerificationFailed();
     }
     if (meta == null) never('Missing metadata from stream');
     const remoteHost = meta.remoteHost;

src/nodes/agent/handlers/NodesConnectionSignalInitial.ts

@@ -48,17 +48,21 @@ class ErrorNodeManagerAuthenticationFailed<T> extends ErrorNodeManager<T> {
   exitCode = sysexits.NOPERM;
 }
 
-class ErrorNodeManagerAuthenticationFailedForward<T> extends ErrorNodes<T> {
+class ErrorNodeManagerAuthenticationFailedForward<
+  T,
+> extends ErrorNodeManager<T> {
   static description = 'Failed to complete forward authentication';
   exitCode = sysexits.USAGE;
 }
 
-class ErrorNodeManagerAuthenticationFailedReverse<T> extends ErrorNodes<T> {
+class ErrorNodeManagerAuthenticationFailedReverse<
+  T,
+> extends ErrorNodeManager<T> {
   static description = 'Failed to complete reverse authentication';
   exitCode = sysexits.USAGE;
 }
 
-class ErrorNodeManagerAuthenticatonTimedOut<T> extends ErrorNodes<T> {
+class ErrorNodeManagerAuthenticationTimedOut<T> extends ErrorNodeManager<T> {
   static description = 'Failed to complete authentication before timing out';
   exitCode = sysexits.USAGE;
 }
@@ -154,7 +158,7 @@ class ErrorNodeConnectionTransportGenericError<
   exitCode = sysexits.USAGE;
 }
 
-class ErrorConnectionNodesEmpty<T> extends ErrorNodeConnection<T> {
+class ErrorNodeConnectionEmpty<T> extends ErrorNodeConnection<T> {
   static description = 'Nodes list to verify against was empty';
   exitCode = sysexits.USAGE;
 }
@@ -256,6 +260,30 @@ class ErrorNodeAuthenticationFailed<T> extends ErrorNodes<T> {
   exitCode = sysexits.NOPERM;
 }
 
+class ErrorNodeAuthenticationInvalidProtocol<T> extends ErrorNodes<T> {
+  static description = 'Invalid protocol used for node authentication';
+  exitCode = sysexits.USAGE;
+}
+
+class ErrorNodeConnectionSignalRequestVerificationFailed<
+  T,
+> extends ErrorNodeConnection<T> {
+  static description = 'Failed to verify request message signature';
+  exitCode = sysexits.UNAVAILABLE;
+}
+
+class ErrorNodeConnectionSignalRelayVerificationFailed<
+  T,
+> extends ErrorNodeConnection<T> {
+  static description = 'Failed to verify relay message signature';
+  exitCode = sysexits.UNAVAILABLE;
+}
+
+class ErrorNodeClaimNetworkVerificationFailed<T> extends ErrorNodes<T> {
+  static description = 'Failed to verify claim network message';
+  exitCode = sysexits.UNAVAILABLE;
+}
+
 export {
   ErrorNodes,
   ErrorNodeManager,
@@ -269,7 +297,7 @@ export {
   ErrorNodeManagerAuthenticationFailed,
   ErrorNodeManagerAuthenticationFailedForward,
   ErrorNodeManagerAuthenticationFailedReverse,
-  ErrorNodeManagerAuthenticatonTimedOut,
+  ErrorNodeManagerAuthenticationTimedOut,
   ErrorNodeGraph,
   ErrorNodeGraphRunning,
   ErrorNodeGraphNotRunning,
@@ -288,7 +316,7 @@ export {
   ErrorNodeConnectionInternalError,
   ErrorNodeConnectionTransportUnknownError,
   ErrorNodeConnectionTransportGenericError,
-  ErrorConnectionNodesEmpty,
+  ErrorNodeConnectionEmpty,
   ErrorNodeConnectionManager,
   ErrorNodeConnectionManagerNotRunning,
   ErrorNodeConnectionManagerStopping,
@@ -304,4 +332,8 @@ export {
   ErrorNodePermissionDenied,
   ErrorNodeLookupNotFound,
   ErrorNodeAuthenticationFailed,
+  ErrorNodeAuthenticationInvalidProtocol,
+  ErrorNodeClaimNetworkVerificationFailed,
+  ErrorNodeConnectionSignalRelayVerificationFailed,
+  ErrorNodeConnectionSignalRequestVerificationFailed,
 };

src/nodes/errors.ts

@@ -1,6 +1,7 @@
+import type { ContextTimed } from '@matrixai/contexts';
 import type { DBTransaction, KeyPath, LevelPath } from '@matrixai/db';
-import type { X509Certificate } from '@peculiar/x509';
 import type { QUICClientCrypto, QUICServerCrypto } from '@matrixai/quic';
+import type { X509Certificate } from '@peculiar/x509';
 import type { Key, Certificate, CertificatePEM } from '../keys/types.js';
 import type { Hostname, Port } from '../network/types.js';
 import type {
@@ -18,7 +19,6 @@ import type {
   NodesAuthenticateConnectionMessageBasicPublic,
   NodesAuthenticateConnectionMessageNone,
 } from './agent/types.js';
-import type { ContextTimed } from '@matrixai/contexts';
 import dns from 'dns';
 import { utils as dbUtils } from '@matrixai/db';
 import { IdInternal } from '@matrixai/id';
@@ -566,7 +566,7 @@ async function verifyServerCertificateChain(
     };
   }
   if (nodeIds.length === 0) {
-    throw new nodesErrors.ErrorConnectionNodesEmpty();
+    throw new nodesErrors.ErrorNodeConnectionEmpty();
   }
   const certChain: Array<Readonly<X509Certificate>> = [];
   for (const certPEM of certPEMChain) {

src/nodes/utils.ts

@@ -547,6 +547,42 @@ async function importFS(fs?: FileSystem): Promise<FileSystem> {
   return fsImported;
 }
 
+/**
+ * Races a promise against a context. If the context is aborted, then the promise
+ * rejects with the reason. Otherwise, the original value is returned upon
+ * resolving.
+ * @param prom The promise to resolve
+ * @param ctx The ctx with which to race the promise against
+ * @returns A promise which resolves to the prom value or errors if ctx aborts
+ */
+async function resultOrAbort<T>(
+  prom: Promise<T>,
+  ctx: ContextTimed,
+): Promise<T> {
+  // Create an abort promise which rejects when ctx is aborted
+  const { p: abortP, rejectP: rejectAbortP } = promise<never>();
+  const abortHandler = () => {
+    rejectAbortP(ctx.signal.reason);
+  };
+  if (ctx.signal.aborted) {
+    abortHandler();
+  } else {
+    ctx.signal.addEventListener('abort', abortHandler, { once: true });
+  }
+
+  // Race the original promise and abortP. If the original promise resolves
+  // first, then it is returned. If the context aborts first, then the
+  // promise is rejected.
+  try {
+    return await Promise.race([prom, abortP]);
+  } catch (e) {
+    Error.captureStackTrace(e);
+    throw e;
+  } finally {
+    ctx.signal.removeEventListener('abort', abortHandler);
+  }
+}
+
 export {
   AsyncFunction,
   GeneratorFunction,
@@ -588,4 +624,5 @@ export {
   yieldMicro,
   setMaxListeners,
   importFS,
+  resultOrAbort,
 };

src/utils/utils.ts

@@ -86,7 +86,7 @@ describe('IdentitiesManager', () => {
       await identitiesManager.getTokens('abc' as ProviderId);
     }).rejects.toThrow(identitiesErrors.ErrorIdentitiesManagerNotRunning);
   });
-  test.only.prop([
+  test.prop([
     identitiesTestUtils.identitiyIdArb,
     identitiesTestUtils.providerTokenArb,
   ])('get, set and unset tokens', async (identityId, providerToken) => {