feat(auth): add Entra ID identity provider integration for Redis client authentication (#2877)

* 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

* feat(auth): auth extensions

Introduces TokenManager and supporting classes to handle token acquisition, automatic
refresh, and updates via identity providers. This foundation enables consistent
authentication token management across different identity provider implementations.

Key additions:
- Add TokenManager to obtain and maintain auth tokens from identity providers
  with automated refresh scheduling based on TTL and configurable thresholds
- Add IdentityProvider interface for token acquisition from auth providers
- Implement Token class for managing token state and TTL tracking
- Include configurable retry mechanism with exponential backoff and jitter
- Add comprehensive test suite covering refresh cycles and error handling

This change establishes the core infrastructure needed for reliable token
lifecycle management across different authentication providers.

* feat(auth): add Entra ID identity provider integration

Introduces Entra ID (former Azure AD) authentication support with multiple authentication flows
and automated token lifecycle management.

Key additions:
- Add EntraIdCredentialsProvider for handling Entra ID authentication flows
- Implement MSALIdentityProvider to integrate with MSAL/EntraID authentication library
- Add support for multiple authentication methods:
  - Managed identities (system and user-assigned)
  - Client credentials with certificate
  - Client credentials with secret
  - Authorization Code flow with PKCE
- Add factory class with builder methods for each authentication flow
- Include sample Express server implementation for Authorization Code flow
- Add comprehensive configuration options for authority and token management

* feat(test-utils): improve cluster testing

- Add support for configuring replica authentication with 'masterauth'
- Allow default client configuration during test cluster creation

This improves the testing framework's flexibility by automatically
configuring replica authentication when '--requirepass' is used and
enabling custom client configurations across cluster nodes.

* feat(auth): add EntraId integration tests

- Add integration tests for token renewal and re-authentication flows
- Update credentials provider to use uniqueId as username instead of account username
- Add test utilities for loading Redis endpoint configurations
- Split TypeScript configs into separate files for samples and integration tests
- Remove `@redis/authx` package and nest it under `@`

packages/entraid/lib/entra-id-credentials-provider-factory.ts

@@ -0,0 +1,371 @@
+import { NetworkError } from '@azure/msal-common';
+import {
+  LogLevel,
+  ManagedIdentityApplication,
+  ManagedIdentityConfiguration,
+  AuthenticationResult,
+  PublicClientApplication,
+  ConfidentialClientApplication, AuthorizationUrlRequest, AuthorizationCodeRequest, CryptoProvider, Configuration, NodeAuthOptions, AccountInfo
+} from '@azure/msal-node';
+import { RetryPolicy, TokenManager, TokenManagerConfig, ReAuthenticationError } from '@redis/client/dist/lib/authx';
+import { EntraidCredentialsProvider } from './entraid-credentials-provider';
+import { MSALIdentityProvider } from './msal-identity-provider';
+
+/**
+ * This class is used to create credentials providers for different types of authentication flows.
+ */
+export class EntraIdCredentialsProviderFactory {
+
+  /**
+   * This method is used to create a ManagedIdentityProvider for both system-assigned and user-assigned managed identities.
+   *
+   * @param params
+   * @param userAssignedClientId For user-assigned managed identities, the developer needs to pass either the client ID,
+   * full resource identifier, or the object ID of the managed identity when creating ManagedIdentityApplication.
+   *
+   */
+  public static createManagedIdentityProvider(
+    params: CredentialParams, userAssignedClientId?: string
+  ): EntraidCredentialsProvider {
+    const config: ManagedIdentityConfiguration = {
+      // For user-assigned identity, include the client ID
+      ...(userAssignedClientId && {
+        managedIdentityIdParams: {
+          userAssignedClientId
+        }
+      }),
+      system: {
+        loggerOptions
+      }
+    };
+
+    const client = new ManagedIdentityApplication(config);
+
+    const idp = new MSALIdentityProvider(
+      () => client.acquireToken({
+        resource: params.scopes?.[0] ?? REDIS_SCOPE,
+        forceRefresh: true
+      }).then(x => x === null ? Promise.reject('Token is null') : x)
+    );
+
+    return new EntraidCredentialsProvider(
+      new TokenManager(idp, params.tokenManagerConfig),
+      idp,
+      { onReAuthenticationError: params.onReAuthenticationError, credentialsMapper: OID_CREDENTIALS_MAPPER }
+    );
+  }
+
+  /**
+   * This method is used to create a credentials provider for system-assigned managed identities.
+   * @param params
+   */
+  static createForSystemAssignedManagedIdentity(
+    params: CredentialParams
+  ): EntraidCredentialsProvider {
+    return this.createManagedIdentityProvider(params);
+  }
+
+  /**
+   * This method is used to create a credentials provider for user-assigned managed identities.
+   * It will include the client ID as the userAssignedClientId in the ManagedIdentityConfiguration.
+   * @param params
+   */
+  static createForUserAssignedManagedIdentity(
+    params: CredentialParams & { userAssignedClientId: string }
+  ): EntraidCredentialsProvider {
+    return this.createManagedIdentityProvider(params, params.userAssignedClientId);
+  }
+
+  static #createForClientCredentials(
+    authConfig: NodeAuthOptions,
+    params: CredentialParams
+  ): EntraidCredentialsProvider {
+    const config: Configuration = {
+      auth: {
+        ...authConfig,
+        authority: this.getAuthority(params.authorityConfig ?? { type: 'default' })
+      },
+      system: {
+        loggerOptions
+      }
+    };
+
+    const client = new ConfidentialClientApplication(config);
+
+    const idp = new MSALIdentityProvider(
+      () => client.acquireTokenByClientCredential({
+        skipCache: true,
+        scopes: params.scopes ?? [REDIS_SCOPE_DEFAULT]
+      }).then(x => x === null ? Promise.reject('Token is null') : x)
+    );
+
+    return new EntraidCredentialsProvider(new TokenManager(idp, params.tokenManagerConfig), idp,
+      {
+        onReAuthenticationError: params.onReAuthenticationError,
+        credentialsMapper: OID_CREDENTIALS_MAPPER
+      });
+  }
+
+  /**
+   * This method is used to create a credentials provider for service principals using certificate.
+   * @param params
+   */
+  static createForClientCredentialsWithCertificate(
+    params: ClientCredentialsWithCertificateParams
+  ): EntraidCredentialsProvider {
+    return this.#createForClientCredentials(
+      {
+        clientId: params.clientId,
+        clientCertificate: params.certificate
+      },
+      params
+    );
+  }
+
+  /**
+   * This method is used to create a credentials provider for service principals using client secret.
+   * @param params
+   */
+  static createForClientCredentials(
+    params: ClientSecretCredentialsParams
+  ): EntraidCredentialsProvider {
+    return this.#createForClientCredentials(
+      {
+        clientId: params.clientId,
+        clientSecret: params.clientSecret
+      },
+      params
+    );
+  }
+
+  /**
+   * This method is used to create a credentials provider for the Authorization Code Flow with PKCE.
+   * @param params
+   */
+  static createForAuthorizationCodeWithPKCE(
+    params: AuthCodePKCEParams
+  ): {
+    getPKCECodes: () => Promise<{
+      verifier: string;
+      challenge: string;
+      challengeMethod: string;
+    }>;
+    getAuthCodeUrl: (
+      pkceCodes: { challenge: string; challengeMethod: string }
+    ) => Promise<string>;
+    createCredentialsProvider: (
+      params: PKCEParams
+    ) => EntraidCredentialsProvider;
+  } {
+
+    const requiredScopes = ['user.read', 'offline_access'];
+    const scopes = [...new Set([...(params.scopes || []), ...requiredScopes])];
+
+    const authFlow = AuthCodeFlowHelper.create({
+      clientId: params.clientId,
+      redirectUri: params.redirectUri,
+      scopes: scopes,
+      authorityConfig: params.authorityConfig
+    });
+
+    return {
+      getPKCECodes: AuthCodeFlowHelper.generatePKCE,
+      getAuthCodeUrl: (pkceCodes) => authFlow.getAuthCodeUrl(pkceCodes),
+      createCredentialsProvider: (pkceParams) => {
+
+        // This is used to store the initial credentials account to be used
+        // for silent token acquisition after the initial token acquisition.
+        let initialCredentialsAccount: AccountInfo | null = null;
+
+        const idp = new MSALIdentityProvider(
+          async () => {
+            if (!initialCredentialsAccount) {
+              let authResult = await authFlow.acquireTokenByCode(pkceParams);
+              initialCredentialsAccount = authResult.account;
+              return authResult;
+            } else {
+              return authFlow.client.acquireTokenSilent({
+                forceRefresh: true,
+                account: initialCredentialsAccount,
+                scopes
+              });
+            }
+
+          }
+        );
+        const tm = new TokenManager(idp, params.tokenManagerConfig);
+        return new EntraidCredentialsProvider(tm, idp, { onReAuthenticationError: params.onReAuthenticationError });
+      }
+    };
+  }
+
+  static getAuthority(config: AuthorityConfig): string {
+    switch (config.type) {
+      case 'multi-tenant':
+        return `https://login.microsoftonline.com/${config.tenantId}`;
+      case 'custom':
+        return config.authorityUrl;
+      case 'default':
+        return 'https://login.microsoftonline.com/common';
+      default:
+        throw new Error('Invalid authority configuration');
+    }
+  }
+
+}
+
+const REDIS_SCOPE_DEFAULT = 'https://redis.azure.com/.default';
+const REDIS_SCOPE = 'https://redis.azure.com'
+
+export type AuthorityConfig =
+  | { type: 'multi-tenant'; tenantId: string }
+  | { type: 'custom'; authorityUrl: string }
+  | { type: 'default' };
+
+export type PKCEParams = {
+  code: string;
+  verifier: string;
+  clientInfo?: string;
+}
+
+export type CredentialParams = {
+  clientId: string;
+  scopes?: string[];
+  authorityConfig?: AuthorityConfig;
+
+  tokenManagerConfig: TokenManagerConfig
+  onReAuthenticationError?: (error: ReAuthenticationError) => void;
+}
+
+export type AuthCodePKCEParams = CredentialParams & {
+  redirectUri: string;
+};
+
+export type ClientSecretCredentialsParams = CredentialParams & {
+  clientSecret: string;
+};
+
+export type ClientCredentialsWithCertificateParams = CredentialParams & {
+  certificate: {
+    thumbprint: string;
+    privateKey: string;
+    x5c?: string;
+  };
+};
+
+const loggerOptions = {
+  loggerCallback(loglevel: LogLevel, message: string, containsPii: boolean) {
+    if (!containsPii) console.log(message);
+  },
+  piiLoggingEnabled: false,
+  logLevel: LogLevel.Error
+}
+
+/**
+ * The most important part of the RetryPolicy is the `isRetryable` function. This function is used to determine if a request should be retried based
+ * on the error returned from the identity provider. The default for is to retry on network errors only.
+ */
+export const DEFAULT_RETRY_POLICY: RetryPolicy = {
+  // currently only retry on network errors
+  isRetryable: (error: unknown) => error instanceof NetworkError,
+  maxAttempts: 10,
+  initialDelayMs: 100,
+  maxDelayMs: 100000,
+  backoffMultiplier: 2,
+  jitterPercentage: 0.1
+
+};
+
+export const DEFAULT_TOKEN_MANAGER_CONFIG: TokenManagerConfig = {
+  retry: DEFAULT_RETRY_POLICY,
+  expirationRefreshRatio: 0.7 // Refresh token when 70% of the token has expired
+}
+
+/**
+ * This class is used to help with the Authorization Code Flow with PKCE.
+ * It provides methods to generate PKCE codes, get the authorization URL, and create the credential provider.
+ */
+export class AuthCodeFlowHelper {
+  private constructor(
+    readonly client: PublicClientApplication,
+    readonly scopes: string[],
+    readonly redirectUri: string
+  ) {}
+
+  async getAuthCodeUrl(pkceCodes: {
+    challenge: string;
+    challengeMethod: string;
+  }): Promise<string> {
+    const authCodeUrlParameters: AuthorizationUrlRequest = {
+      scopes: this.scopes,
+      redirectUri: this.redirectUri,
+      codeChallenge: pkceCodes.challenge,
+      codeChallengeMethod: pkceCodes.challengeMethod
+    };
+
+    return this.client.getAuthCodeUrl(authCodeUrlParameters);
+  }
+
+  async acquireTokenByCode(params: PKCEParams): Promise<AuthenticationResult> {
+    const tokenRequest: AuthorizationCodeRequest = {
+      code: params.code,
+      scopes: this.scopes,
+      redirectUri: this.redirectUri,
+      codeVerifier: params.verifier,
+      clientInfo: params.clientInfo
+    };
+
+    return this.client.acquireTokenByCode(tokenRequest);
+  }
+
+  static async generatePKCE(): Promise<{
+    verifier: string;
+    challenge: string;
+    challengeMethod: string;
+  }> {
+    const cryptoProvider = new CryptoProvider();
+    const { verifier, challenge } = await cryptoProvider.generatePkceCodes();
+    return {
+      verifier,
+      challenge,
+      challengeMethod: 'S256'
+    };
+  }
+
+  static create(params: {
+    clientId: string;
+    redirectUri: string;
+    scopes?: string[];
+    authorityConfig?: AuthorityConfig;
+  }): AuthCodeFlowHelper {
+    const config = {
+      auth: {
+        clientId: params.clientId,
+        authority: EntraIdCredentialsProviderFactory.getAuthority(params.authorityConfig ?? { type: 'default' })
+      },
+      system: {
+        loggerOptions
+      }
+    };
+
+    return new AuthCodeFlowHelper(
+      new PublicClientApplication(config),
+      params.scopes ?? ['user.read'],
+      params.redirectUri
+    );
+  }
+}
+
+const OID_CREDENTIALS_MAPPER = (token: AuthenticationResult) => {
+
+  // Client credentials flow is app-only authentication (no user context),
+  // so only access token is provided without user-specific claims (uniqueId, idToken, ...)
+  // this means that we need to extract the oid from the access token manually
+  const accessToken = JSON.parse(Buffer.from(token.accessToken.split('.')[1], 'base64').toString());
+
+  return ({
+    username: accessToken.oid,
+    password: token.accessToken
+  })
+
+}

packages/entraid/lib/entraid-credentials-provider.spec.ts

@@ -0,0 +1,199 @@
+import { AuthenticationResult } from '@azure/msal-node';
+import { IdentityProvider, TokenManager, TokenResponse, BasicAuth } from '@redis/client/dist/lib/authx';
+import { EntraidCredentialsProvider } from './entraid-credentials-provider';
+import { setTimeout } from 'timers/promises';
+import { strict as assert } from 'node:assert';
+import { GLOBAL, testUtils } from './test-utils'
+
+
+describe('EntraID authentication in cluster mode', () => {
+
+  testUtils.testWithCluster('sendCommand', async cluster => {
+    assert.equal(
+      await cluster.sendCommand(undefined, true, ['PING']),
+      'PONG'
+    );
+  }, GLOBAL.CLUSTERS.PASSWORD_WITH_REPLICAS);
+})
+
+describe('EntraID CredentialsProvider Subscription Behavior', () => {
+
+  it('should properly handle token refresh sequence for multiple subscribers', async () => {
+    const networkDelay = 20;
+    const tokenTTL = 100;
+    const refreshRatio = 0.5; // Refresh at 50% of TTL
+
+    const idp = new SequenceEntraIDProvider(tokenTTL, networkDelay);
+    const tokenManager = new TokenManager<AuthenticationResult>(idp, {
+      expirationRefreshRatio: refreshRatio
+    });
+    const entraid = new EntraidCredentialsProvider(tokenManager, idp);
+
+    // Create two initial subscribers
+    const subscriber1 = new TestSubscriber('subscriber1');
+    const subscriber2 = new TestSubscriber('subscriber2');
+
+    assert.equal(entraid.hasActiveSubscriptions(), false, 'There should be no active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 0, 'There should be 0 subscriptions');
+
+    // Start the first two subscriptions almost simultaneously
+    const [sub1Initial, sub2Initial] = await Promise.all([
+      entraid.subscribe(subscriber1),
+      entraid.subscribe(subscriber2)]
+    );
+
+    assertCredentials(sub1Initial[0], 'initial-token', 'Subscriber 1 should receive initial token');
+    assertCredentials(sub2Initial[0], 'initial-token', 'Subscriber 2 should receive initial token');
+
+    assert.equal(entraid.hasActiveSubscriptions(), true, 'There should be active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 2, 'There should be 2 subscriptions');
+
+    // add a third subscriber after a very short delay
+    const subscriber3 = new TestSubscriber('subscriber3');
+    await setTimeout(1);
+    const sub3Initial = await entraid.subscribe(subscriber3)
+
+    assert.equal(entraid.hasActiveSubscriptions(), true, 'There should be active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 3, 'There should be 3 subscriptions');
+
+    // make sure the third subscriber gets the initial token as well
+    assertCredentials(sub3Initial[0], 'initial-token', 'Subscriber 3 should receive initial token');
+
+    // Wait for first refresh (50% of TTL + network delay + small buffer)
+    await setTimeout((tokenTTL * refreshRatio) + networkDelay + 15);
+
+    // All 3 subscribers should receive refresh-token-1
+    assertCredentials(subscriber1.credentials[0], 'refresh-token-1', 'Subscriber 1 should receive first refresh token');
+    assertCredentials(subscriber2.credentials[0], 'refresh-token-1', 'Subscriber 2 should receive first refresh token');
+    assertCredentials(subscriber3.credentials[0], 'refresh-token-1', 'Subscriber 3 should receive first refresh token');
+
+    // Add a late subscriber - should immediately get refresh-token-1
+    const subscriber4 = new TestSubscriber('subscriber4');
+    const sub4Initial = await entraid.subscribe(subscriber4);
+
+    assert.equal(entraid.hasActiveSubscriptions(), true, 'There should be active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 4, 'There should be 4 subscriptions');
+
+    assertCredentials(sub4Initial[0], 'refresh-token-1', 'Late subscriber should receive refresh-token-1');
+
+    // Wait for second refresh
+    await setTimeout((tokenTTL * refreshRatio) + networkDelay + 15);
+
+    assertCredentials(subscriber1.credentials[1], 'refresh-token-2', 'Subscriber 1 should receive second refresh token');
+    assertCredentials(subscriber2.credentials[1], 'refresh-token-2', 'Subscriber 2 should receive second refresh token');
+    assertCredentials(subscriber3.credentials[1], 'refresh-token-2', 'Subscriber 3 should receive second refresh token');
+
+    assertCredentials(subscriber4.credentials[0], 'refresh-token-2', 'Subscriber 4 should receive second refresh token');
+
+    // Verify refreshes happen after minimum expected time
+    const minimumRefreshInterval = tokenTTL * 0.4; // 40% of TTL as safety margin
+
+    verifyRefreshTiming(subscriber1, minimumRefreshInterval);
+    verifyRefreshTiming(subscriber2, minimumRefreshInterval);
+    verifyRefreshTiming(subscriber3, minimumRefreshInterval);
+    verifyRefreshTiming(subscriber4, minimumRefreshInterval);
+
+    // Cleanup
+
+    assert.equal(tokenManager.isRunning(), true);
+    sub1Initial[1].dispose();
+    sub2Initial[1].dispose();
+    sub3Initial[1].dispose();
+    assert.equal(entraid.hasActiveSubscriptions(), true, 'There should be active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 1, 'There should be 1 subscriptions');
+    sub4Initial[1].dispose();
+    assert.equal(entraid.hasActiveSubscriptions(), false, 'There should be no active subscriptions');
+    assert.equal(entraid.getSubscriptionsCount(), 0, 'There should be 0 subscriptions');
+    assert.equal(tokenManager.isRunning(), false)
+  });
+
+  const verifyRefreshTiming = (
+    subscriber: TestSubscriber,
+    expectedMinimumInterval: number,
+    message?: string
+  ) => {
+    const intervals = [];
+    for (let i = 1; i < subscriber.timestamps.length; i++) {
+      intervals.push(subscriber.timestamps[i] - subscriber.timestamps[i - 1]);
+    }
+
+    intervals.forEach((interval, index) => {
+      assert.ok(
+        interval > expectedMinimumInterval,
+        message || `Refresh ${index + 1} for ${subscriber.name} should happen after minimum interval of ${expectedMinimumInterval}ms`
+      );
+    });
+  };
+
+  class SequenceEntraIDProvider implements IdentityProvider<AuthenticationResult> {
+    private currentIndex = 0;
+
+    constructor(
+      private readonly tokenTTL: number = 100,
+      private tokenDeliveryDelayMs: number = 0,
+      private readonly tokenSequence: AuthenticationResult[] = [
+        {
+          accessToken: 'initial-token',
+          uniqueId: 'test-user'
+        } as AuthenticationResult,
+        {
+          accessToken: 'refresh-token-1',
+          uniqueId: 'test-user'
+        } as AuthenticationResult,
+        {
+          accessToken: 'refresh-token-2',
+          uniqueId: 'test-user'
+        } as AuthenticationResult
+      ]
+    ) {}
+
+    setTokenDeliveryDelay(delayMs: number): void {
+      this.tokenDeliveryDelayMs = delayMs;
+    }
+
+    async requestToken(): Promise<TokenResponse<AuthenticationResult>> {
+      if (this.tokenDeliveryDelayMs > 0) {
+        await setTimeout(this.tokenDeliveryDelayMs);
+      }
+
+      if (this.currentIndex >= this.tokenSequence.length) {
+        throw new Error('No more tokens in sequence');
+      }
+
+      return {
+        token: this.tokenSequence[this.currentIndex++],
+        ttlMs: this.tokenTTL
+      };
+    }
+  }
+
+  class TestSubscriber {
+    public readonly credentials: Array<BasicAuth> = [];
+    public readonly errors: Error[] = [];
+    public readonly timestamps: number[] = [];
+
+    constructor(public readonly name: string = 'unnamed') {}
+
+    onNext = (creds: BasicAuth) => {
+      this.credentials.push(creds);
+      this.timestamps.push(Date.now());
+    }
+
+    onError = (error: Error) => {
+      this.errors.push(error);
+    }
+  }
+
+  /**
+   * Assert that the actual credentials match the expected token
+   * @param actual
+   * @param expectedToken
+   * @param message
+   */
+  const assertCredentials = (actual: BasicAuth, expectedToken: string, message: string) => {
+    assert.deepEqual(actual, {
+      username: 'test-user',
+      password: expectedToken
+    }, message);
+  };
+});

packages/entraid/lib/entraid-credentials-provider.ts

@@ -0,0 +1,140 @@
+import { AuthenticationResult } from '@azure/msal-common/node';
+import {
+  BasicAuth, StreamingCredentialsProvider, IdentityProvider, TokenManager,
+  ReAuthenticationError, StreamingCredentialsListener, IDPError, Token, Disposable
+} from '@redis/client/dist/lib/authx';
+
+/**
+ * A streaming credentials provider that uses the Entraid identity provider to provide credentials.
+ * Please use one of the factory functions in `entraid-credetfactories.ts` to create an instance of this class for the different
+ * type of authentication flows.
+ */
+export class EntraidCredentialsProvider implements StreamingCredentialsProvider {
+  readonly type = 'streaming-credentials-provider';
+
+  readonly #listeners: Set<StreamingCredentialsListener<BasicAuth>> = new Set();
+
+  #tokenManagerDisposable: Disposable | null = null;
+  #isStarting: boolean = false;
+
+  #pendingSubscribers: Array<{
+    resolve: (value: [BasicAuth, Disposable]) => void;
+    reject: (error: Error) => void;
+    pendingListener: StreamingCredentialsListener<BasicAuth>;
+  }> = [];
+
+  constructor(
+    public readonly tokenManager: TokenManager<AuthenticationResult>,
+    public readonly idp: IdentityProvider<AuthenticationResult>,
+    private readonly options: {
+      onReAuthenticationError?: (error: ReAuthenticationError) => void;
+      credentialsMapper?: (token: AuthenticationResult) => BasicAuth;
+      onRetryableError?: (error: string) => void;
+    } = {}
+  ) {
+    this.onReAuthenticationError = options.onReAuthenticationError ?? DEFAULT_ERROR_HANDLER;
+    this.#credentialsMapper = options.credentialsMapper ?? DEFAULT_CREDENTIALS_MAPPER;
+  }
+
+  async subscribe(
+    listener: StreamingCredentialsListener<BasicAuth>
+  ): Promise<[BasicAuth, Disposable]> {
+
+    const currentToken = this.tokenManager.getCurrentToken();
+
+    if (currentToken) {
+      return [this.#credentialsMapper(currentToken.value), this.#createDisposable(listener)];
+    }
+
+    if (this.#isStarting) {
+      return new Promise((resolve, reject) => {
+        this.#pendingSubscribers.push({ resolve, reject, pendingListener: listener });
+      });
+    }
+
+    this.#isStarting = true;
+    try {
+      const initialToken = await this.#startTokenManagerAndObtainInitialToken();
+
+      this.#pendingSubscribers.forEach(({ resolve, pendingListener }) => {
+        resolve([this.#credentialsMapper(initialToken.value), this.#createDisposable(pendingListener)]);
+      });
+      this.#pendingSubscribers = [];
+
+      return [this.#credentialsMapper(initialToken.value), this.#createDisposable(listener)];
+    } finally {
+      this.#isStarting = false;
+    }
+  }
+
+  onReAuthenticationError: (error: ReAuthenticationError) => void;
+
+  #credentialsMapper: (token: AuthenticationResult) => BasicAuth;
+
+  #createTokenManagerListener(subscribers: Set<StreamingCredentialsListener<BasicAuth>>) {
+    return {
+      onError: (error: IDPError): void => {
+        if (!error.isRetryable) {
+          subscribers.forEach(listener => listener.onError(error));
+        } else {
+          this.options.onRetryableError?.(error.message);
+        }
+      },
+      onNext: (token: { value: AuthenticationResult }): void => {
+        const credentials = this.#credentialsMapper(token.value);
+        subscribers.forEach(listener => listener.onNext(credentials));
+      }
+    };
+  }
+
+  #createDisposable(listener: StreamingCredentialsListener<BasicAuth>): Disposable {
+    this.#listeners.add(listener);
+
+    return {
+      dispose: () => {
+        this.#listeners.delete(listener);
+        if (this.#listeners.size === 0 && this.#tokenManagerDisposable) {
+          this.#tokenManagerDisposable.dispose();
+          this.#tokenManagerDisposable = null;
+        }
+      }
+    };
+  }
+
+  async #startTokenManagerAndObtainInitialToken(): Promise<Token<AuthenticationResult>> {
+    const initialResponse = await this.idp.requestToken();
+    const token = this.tokenManager.wrapAndSetCurrentToken(initialResponse.token, initialResponse.ttlMs);
+
+    this.#tokenManagerDisposable = this.tokenManager.start(
+      this.#createTokenManagerListener(this.#listeners),
+      this.tokenManager.calculateRefreshTime(token)
+    );
+    return token;
+  }
+
+  public hasActiveSubscriptions(): boolean {
+    return this.#tokenManagerDisposable !== null && this.#listeners.size > 0;
+  }
+
+  public getSubscriptionsCount(): number {
+    return this.#listeners.size;
+  }
+
+  public getTokenManager() {
+    return this.tokenManager;
+  }
+
+  public getCurrentCredentials(): BasicAuth | null {
+    const currentToken = this.tokenManager.getCurrentToken();
+    return currentToken ? this.#credentialsMapper(currentToken.value) : null;
+  }
+
+}
+
+const DEFAULT_CREDENTIALS_MAPPER = (token: AuthenticationResult): BasicAuth => ({
+  username: token.uniqueId,
+  password: token.accessToken
+});
+
+const DEFAULT_ERROR_HANDLER = (error: ReAuthenticationError) =>
+  console.error('ReAuthenticationError', error);

packages/entraid/lib/index.ts

@@ -0,0 +1,3 @@
+export * from './entra-id-credentials-provider-factory';
+export * from './entraid-credentials-provider';
+export * from './msal-identity-provider';

packages/entraid/lib/msal-identity-provider.ts

@@ -0,0 +1,31 @@
+import {
+  AuthenticationResult
+} from '@azure/msal-node';
+import { IdentityProvider, TokenResponse } from '@redis/client/dist/lib/authx';
+
+export class MSALIdentityProvider implements IdentityProvider<AuthenticationResult> {
+  private readonly getToken: () => Promise<AuthenticationResult>;
+
+  constructor(getToken: () => Promise<AuthenticationResult>) {
+    this.getToken = getToken;
+  }
+
+  async requestToken(): Promise<TokenResponse<AuthenticationResult>> {
+    try {
+      const result = await this.getToken();
+
+      if (!result?.accessToken || !result?.expiresOn) {
+        throw new Error('Invalid token response');
+      }
+      return {
+        token: result,
+        ttlMs: result.expiresOn.getTime() - Date.now()
+      };
+    } catch (error) {
+      throw error;
+    }
+  }
+
+}
+
+

packages/entraid/lib/test-utils.ts

@@ -0,0 +1,46 @@
+import { AuthenticationResult } from '@azure/msal-node';
+import { IdentityProvider, StreamingCredentialsProvider, TokenManager, TokenResponse } from '@redis/client/dist/lib/authx';
+import TestUtils from '@redis/test-utils';
+import { EntraidCredentialsProvider } from './entraid-credentials-provider';
+
+export const testUtils = new TestUtils({
+  dockerImageName: 'redis/redis-stack',
+  dockerImageVersionArgument: 'redis-version',
+  defaultDockerVersion: '7.4.0-v1'
+});
+
+const DEBUG_MODE_ARGS = testUtils.isVersionGreaterThan([7]) ?
+  ['--enable-debug-command', 'yes'] :
+  [];
+
+const idp: IdentityProvider<AuthenticationResult> = {
+  requestToken(): Promise<TokenResponse<AuthenticationResult>> {
+    // @ts-ignore
+    return Promise.resolve({
+      ttlMs: 100000,
+      token: {
+        accessToken: 'password'
+      }
+    })
+  }
+}
+
+const tokenManager = new TokenManager<AuthenticationResult>(idp, { expirationRefreshRatio: 0.8 });
+const entraIdCredentialsProvider: StreamingCredentialsProvider = new EntraidCredentialsProvider(tokenManager, idp)
+
+const PASSWORD_WITH_REPLICAS = {
+  serverArguments: ['--requirepass', 'password', ...DEBUG_MODE_ARGS],
+  numberOfMasters: 2,
+  numberOfReplicas: 1,
+  clusterConfiguration: {
+    defaults: {
+      credentialsProvider: entraIdCredentialsProvider
+    }
+  }
+}
+
+export const GLOBAL = {
+  CLUSTERS: {
+    PASSWORD_WITH_REPLICAS
+  }
+}