feat(auth): refactor authentication mechanism to use CredentialsProvider

- Introduce new credential providers: AsyncCredentialsProvider, StreamingCredentialsProvider
- Update client handshake process to use the new CredentialsProviders and to support async credentials fetch / credentials refresh
- Internal conversion of username/password to a CredentialsProvider
- Modify URL parsing to accommodate the new authentication structure
- Tests

Author: bobymicroby

packages/client/lib/client/authx/credentials-provider.ts

@@ -0,0 +1,107 @@
+/**
+ * Provides credentials asynchronously.
+ */
+export interface AsyncCredentialsProvider {
+  readonly type: 'async-credentials-provider';
+  credentials: () => Promise<BasicAuth>
+}
+
+/**
+ * Provides credentials asynchronously with support for continuous updates via a subscription model.
+ * This is useful for environments where credentials are frequently rotated or updated or can be revoked.
+ */
+export interface StreamingCredentialsProvider {
+  readonly type: 'streaming-credentials-provider';
+
+  /**
+   * Provides initial credentials and subscribes to subsequent updates. This is used internally by the node-redis client
+   * to handle credential rotation and re-authentication.
+   *
+   * Note: The node-redis client manages the subscription lifecycle automatically. Users only need to implement
+   * onReAuthenticationError if they want to be notified about authentication failures.
+   *
+   * Error handling:
+   * - Errors received via onError indicate a fatal issue with the credentials stream
+   * - The stream is automatically closed(disposed) when onError occurs
+   * - onError typically mean the provider failed to fetch new credentials after retrying
+   *
+   * @example
+   * ```ts
+   * const provider = getStreamingProvider();
+   * const [initialCredentials, disposable] = await provider.subscribe({
+   *   onNext: (newCredentials) => {
+   *     // Handle credential update
+   *   },
+   *   onError: (error) => {
+   *     // Handle fatal stream error
+   *   }
+   * });
+   *
+   * @param listener - Callbacks to handle credential updates and errors
+   * @returns A Promise resolving to [initial credentials, cleanup function]
+   */
+  subscribe: (listener: StreamingCredentialsListener<BasicAuth>) => Promise<[BasicAuth, Disposable]>
+
+  /**
+   * Called when authentication fails or credentials cannot be renewed in time.
+   * Implement this to handle authentication errors in your application.
+   *
+   * @param error - Either a CredentialsError (invalid/expired credentials) or
+   *                UnableToObtainNewCredentialsError (failed to fetch new credentials on time)
+   */
+  onReAuthenticationError: (error: ReAuthenticationError) => void;
+
+}
+
+/**
+ * Type representing basic authentication credentials.
+ */
+export type BasicAuth = { username?: string, password?: string }
+
+/**
+ * Callback to handle credential updates and errors.
+ */
+export type StreamingCredentialsListener<T> = {
+  onNext: (credentials: T) => void;
+  onError: (e: Error) => void;
+}
+
+/**
+ * Disposable is an interface for objects that hold resources that should be released when they are no longer needed.
+ */
+export type Disposable = {
+  dispose: () => void;
+}
+
+/**
+ * Providers that can supply authentication credentials
+ */
+export type CredentialsProvider = AsyncCredentialsProvider | StreamingCredentialsProvider
+
+/**
+ * Errors that can occur during re-authentication.
+ */
+export type ReAuthenticationError = CredentialsError | UnableToObtainNewCredentialsError
+
+/**
+ * Thrown when re-authentication fails with provided credentials .
+ * e.g. when the credentials are invalid, expired or revoked.
+ *
+ */
+export class CredentialsError extends Error {
+  constructor(message: string) {
+    super(`Re-authentication with latest credentials failed: ${message}`);
+    this.name = 'CredentialsError';
+  }
+
+}
+
+/**
+ * Thrown when new credentials cannot be obtained before current ones expire
+ */
+export class UnableToObtainNewCredentialsError extends Error {
+  constructor(message: string) {
+    super(`Unable to obtain new credentials : ${message}`);
+    this.name = 'UnableToObtainNewCredentialsError';
+  }
+}

packages/client/lib/client/index.spec.ts

@@ -1,6 +1,6 @@
 import { strict as assert } from 'node:assert';
 import testUtils, { GLOBAL, waitTillBeenCalled } from '../test-utils';
-import RedisClient, { RedisClientType } from '.';
+import RedisClient, { RedisClientOptions, RedisClientType } from '.';
 import { AbortError, ClientClosedError, ClientOfflineError, ConnectionTimeoutError, DisconnectsClientError, ErrorReply, MultiErrorReply, SocketClosedUnexpectedlyError, WatchError } from '../errors';
 import { defineScript } from '../lua-script';
 import { spy } from 'sinon';
@@ -25,36 +25,87 @@ export const SQUARE_SCRIPT = defineScript({
 
 describe('Client', () => {
   describe('parseURL', () => {
-    it('redis://user:secret@localhost:6379/0', () => {
-      assert.deepEqual(
-        RedisClient.parseURL('redis://user:secret@localhost:6379/0'),
-        {
-          socket: {
-            host: 'localhost',
-            port: 6379
-          },
-          username: 'user',
-          password: 'secret',
-          database: 0
+    it('redis://user:secret@localhost:6379/0', async () => {
+      const result = RedisClient.parseURL('redis://user:secret@localhost:6379/0');
+      const expected : RedisClientOptions = {
+        socket: {
+          host: 'localhost',
+          port: 6379
+        },
+        username: 'user',
+        password: 'secret',
+        database: 0,
+        credentialsProvider: {
+          type: 'async-credentials-provider',
+          credentials: async () => ({
+            password: 'secret',
+            username: 'user'
+          })
         }
-      );
+      };
+
+      // Compare everything except the credentials function
+      const { credentialsProvider: resultCredProvider, ...resultRest } = result;
+      const { credentialsProvider: expectedCredProvider, ...expectedRest } = expected;
+
+      // Compare non-function properties
+      assert.deepEqual(resultRest, expectedRest);
+
+      if(result.credentialsProvider.type === 'async-credentials-provider'
+        && expected.credentialsProvider.type === 'async-credentials-provider') {
+
+        // Compare the actual output of the credentials functions
+        const resultCreds = await result.credentialsProvider.credentials();
+        const expectedCreds = await expected.credentialsProvider.credentials();
+        assert.deepEqual(resultCreds, expectedCreds);
+      } else {
+        assert.fail('Credentials provider type mismatch');
+      }
+
+
     });
 
-    it('rediss://user:secret@localhost:6379/0', () => {
-      assert.deepEqual(
-        RedisClient.parseURL('rediss://user:secret@localhost:6379/0'),
-        {
-          socket: {
-            host: 'localhost',
-            port: 6379,
-            tls: true
-          },
-          username: 'user',
-          password: 'secret',
-          database: 0
+    it('rediss://user:secret@localhost:6379/0', async () => {
+      const result = RedisClient.parseURL('rediss://user:secret@localhost:6379/0');
+      const expected: RedisClientOptions = {
+        socket: {
+          host: 'localhost',
+          port: 6379,
+          tls: true
+        },
+        username: 'user',
+        password: 'secret',
+        database: 0,
+        credentialsProvider: {
+          credentials: async () => ({
+            password: 'secret',
+            username: 'user'
+          }),
+          type: 'async-credentials-provider'
         }
-      );
-    });
+      };
+
+      // Compare everything except the credentials function
+      const { credentialsProvider: resultCredProvider, ...resultRest } = result;
+      const { credentialsProvider: expectedCredProvider, ...expectedRest } = expected;
+
+      // Compare non-function properties
+      assert.deepEqual(resultRest, expectedRest);
+      assert.equal(resultCredProvider.type, expectedCredProvider.type);
+
+      if (result.credentialsProvider.type === 'async-credentials-provider' &&
+        expected.credentialsProvider.type === 'async-credentials-provider') {
+
+        // Compare the actual output of the credentials functions
+        const resultCreds = await result.credentialsProvider.credentials();
+        const expectedCreds = await expected.credentialsProvider.credentials();
+        assert.deepEqual(resultCreds, expectedCreds);
+
+      } else {
+        assert.fail('Credentials provider type mismatch');
+      }
+
+    })
 
     it('Invalid protocol', () => {
       assert.throws(
@@ -90,6 +141,21 @@ describe('Client', () => {
       );
     }, GLOBAL.SERVERS.PASSWORD);
 
+    testUtils.testWithClient('Client can authenticate asynchronously ', async client => {
+      assert.equal(
+        await client.ping(),
+        'PONG'
+      );
+    }, GLOBAL.SERVERS.ASYNC_BASIC_AUTH);
+
+    testUtils.testWithClient('Client can authenticate using the streaming credentials provider for initial token acquisition',
+      async client => {
+      assert.equal(
+        await client.ping(),
+        'PONG'
+      );
+    }, GLOBAL.SERVERS.STREAMING_AUTH);
+
     testUtils.testWithClient('should execute AUTH before SELECT', async client => {
       assert.equal(
         (await client.clientInfo()).db,
@@ -294,6 +360,7 @@ describe('Client', () => {
           assert.equal(err.replies.length, 2);
           assert.deepEqual(err.errorIndexes, [1]);
           assert.ok(err.replies[1] instanceof ErrorReply);
+          // @ts-ignore TS2802
           assert.deepEqual([...err.errors()], [err.replies[1]]);
           return true;
         }