[core-amqp][event-hubs] add support for NamedKeyCredential and SASCre…

…dential (#14423)

Adds support for `NamedKeyCredential` and `SASCredential` to `@azure/event-hubs`.

## Background

Event Hubs supports using token-based AAD auth via the `@azure/identity` credentials, and both "shared access key" and "shared access signature" via the connection string.

This change allows the `EventHubConsumerClient` and `EventHubProducerClient` to support both "shared access key" and "shared access signature" methods of auth via credential objects defined by `@azure/core-auth`.

## It already supports connection strings, so why add 2 more credential types?

Both the `AzureNamedKeyCredential` and `AzureSASCredential` classes support rotation by calling their respective `update` methods. Since connection strings can't be rotated within a client, the user would need to close their existing clients (which may be in the process of sending or receiving events!) and create new ones if their keys were rotated.

With this change, the user only needs to call `update` on the credential they passed to their Event Hubs client, and the client will automatically use the new values.

## Why is this change in core-amqp as well?

Service Bus and Event Hubs both need these changes. Note that the APIs added to core-amqp have the `@hidden` tag so they won't show up in documentation.

## TODO

- [x] verify that the SDK retries 'unauthorized' errors. This, and the token expiry, will determine when the clients attempt to use the rotated credentials.

sdk/core/core-amqp/package.json

@@ -70,13 +70,12 @@
   },
   "dependencies": {
     "@azure/abort-controller": "^1.0.0",
+    "@azure/core-auth": "^1.2.0",
     "@azure/logger": "^1.0.0",
     "@types/async-lock": "^1.1.0",
-    "@types/is-buffer": "^2.0.0",
     "async-lock": "^1.1.3",
     "buffer": "^5.2.1",
     "events": "^3.0.0",
-    "is-buffer": "^2.0.3",
     "jssha": "^3.1.0",
     "process": "^0.11.10",
     "rhea": "^1.0.24",

sdk/core/core-amqp/review/core-amqp.api.md

@@ -5,15 +5,18 @@
 ```ts
 
 import { AbortSignalLike } from '@azure/abort-controller';
+import { AccessToken } from '@azure/core-auth';
 import { AmqpError } from 'rhea-promise';
 import AsyncLock from 'async-lock';
 import { Connection } from 'rhea-promise';
 import { Message } from 'rhea-promise';
 import { MessageHeader } from 'rhea-promise';
 import { MessageProperties } from 'rhea-promise';
+import { NamedKeyCredential } from '@azure/core-auth';
 import { Receiver } from 'rhea-promise';
 import { ReceiverOptions } from 'rhea-promise';
 import { ReqResLink } from 'rhea-promise';
+import { SASCredential } from '@azure/core-auth';
 import { Sender } from 'rhea-promise';
 import { SenderOptions } from 'rhea-promise';
 import { Session } from 'rhea-promise';
@@ -338,6 +341,14 @@ export interface CreateConnectionContextBaseParameters {
     operationTimeoutInMs?: number;
 }
 
+// @public
+export function createSasTokenProvider(data: {
+    sharedAccessKeyName: string;
+    sharedAccessKey: string;
+} | {
+    sharedAccessSignature: string;
+} | NamedKeyCredential | SASCredential): SasTokenProvider;
+
 // @public
 export const defaultLock: AsyncLock;
 
@@ -396,6 +407,9 @@ export enum ErrorNameConditionMapper {
 // @public
 export function isMessagingError(error: Error | MessagingError): error is MessagingError;
 
+// @public
+export function isSasTokenProvider(thing: unknown): thing is SasTokenProvider;
+
 // @public
 export function isSystemError(err: unknown): err is NetworkSystemError;
 
@@ -517,6 +531,12 @@ export interface RetryOptions {
     timeoutInMs?: number;
 }
 
+// @public
+export interface SasTokenProvider {
+    getToken(audience: string): AccessToken;
+    isSasTokenProvider: true;
+}
+
 // @public
 export interface SendRequestOptions {
     abortSignal?: AbortSignalLike;

sdk/core/core-amqp/src/auth/tokenProvider.ts

@@ -0,0 +1,123 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import {
+  AccessToken,
+  NamedKeyCredential,
+  SASCredential,
+  isNamedKeyCredential,
+  isSASCredential
+} from "@azure/core-auth";
+import jssha from "jssha";
+import { isObjectWithProperties } from "../util/typeGuards";
+
+/**
+ * A SasTokenProvider provides an alternative to TokenCredential for providing an `AccessToken`.
+ * @hidden
+ */
+export interface SasTokenProvider {
+  /**
+   * Property used to distinguish SasTokenProvider from TokenCredential.
+   */
+  isSasTokenProvider: true;
+  /**
+   * Gets the token provided by this provider.
+   *
+   * This method is called automatically by Azure SDK client libraries.
+   *
+   * @param audience - The audience for which the token is desired.
+   */
+  getToken(audience: string): AccessToken;
+}
+
+/**
+ * Creates a token provider from the provided shared access data.
+ * @param data - The sharedAccessKeyName/sharedAccessKey pair or the sharedAccessSignature.
+ * @hidden
+ */
+export function createSasTokenProvider(
+  data:
+    | { sharedAccessKeyName: string; sharedAccessKey: string }
+    | { sharedAccessSignature: string }
+    | NamedKeyCredential
+    | SASCredential
+): SasTokenProvider {
+  if (isNamedKeyCredential(data) || isSASCredential(data)) {
+    return new SasTokenProviderImpl(data);
+  } else if (isObjectWithProperties(data, ["sharedAccessKeyName", "sharedAccessKey"])) {
+    return new SasTokenProviderImpl({ name: data.sharedAccessKeyName, key: data.sharedAccessKey });
+  } else {
+    return new SasTokenProviderImpl({ signature: data.sharedAccessSignature });
+  }
+}
+
+/**
+ * A TokenProvider that generates a Sas token:
+ * `SharedAccessSignature sr=<resource>&sig=<signature>&se=<expiry>&skn=<keyname>`
+ *
+ * @internal
+ */
+export class SasTokenProviderImpl implements SasTokenProvider {
+  /**
+   * Property used to distinguish TokenProvider from TokenCredential.
+   */
+  get isSasTokenProvider(): true {
+    return true;
+  }
+
+  /**
+   * The SASCredential containing the key name and secret key value.
+   */
+  private _credential: SASCredential | NamedKeyCredential;
+
+  /**
+   * Initializes a new instance of SasTokenProvider
+   * @param credential - The source `NamedKeyCredential` or `SASCredential`.
+   */
+  constructor(credential: SASCredential | NamedKeyCredential) {
+    this._credential = credential;
+  }
+
+  /**
+   * Gets the sas token for the specified audience
+   * @param audience - The audience for which the token is desired.
+   */
+  getToken(audience: string): AccessToken {
+    if (isNamedKeyCredential(this._credential)) {
+      return createToken(
+        this._credential.name,
+        this._credential.key,
+        Math.floor(Date.now() / 1000) + 3600,
+        audience
+      );
+    } else {
+      return {
+        token: this._credential.signature,
+        expiresOnTimestamp: 0
+      };
+    }
+  }
+}
+
+/**
+ * Creates the sas token based on the provided information.
+ * @param keyName - The shared access key name.
+ * @param key - The shared access key.
+ * @param expiry - The time period in unix time after which the token will expire.
+ * @param audience - The audience for which the token is desired.
+ * @internal
+ */
+function createToken(keyName: string, key: string, expiry: number, audience: string): AccessToken {
+  audience = encodeURIComponent(audience);
+  keyName = encodeURIComponent(keyName);
+  const stringToSign = audience + "\n" + expiry;
+
+  const shaObj = new jssha("SHA-256", "TEXT");
+  shaObj.setHMACKey(key, "TEXT");
+  shaObj.update(stringToSign);
+  const sig = encodeURIComponent(shaObj.getHMAC("B64"));
+  return {
+    token: `SharedAccessSignature sr=${audience}&sig=${sig}&se=${expiry}&skn=${keyName}`,
+    expiresOnTimestamp: expiry
+  };
+}

sdk/core/core-amqp/src/errors.ts

@@ -554,6 +554,10 @@ export const retryableErrors: string[] = [
   "ServiceUnavailableError",
   "OperationCancelledError",
 
+  // The service may throw UnauthorizedError if credentials have been rotated.
+  // Attempt to retry in case the user has also rotated their credentials.
+  "UnauthorizedError",
+
   // OperationTimeoutError occurs when the service fails to respond within a given timeframe.
   // Since reasons for such failures can be transient, this is treated as a retryable error.
   "OperationTimeoutError",

sdk/core/core-amqp/src/index.ts

@@ -40,3 +40,4 @@ export {
 } from "./util/utils";
 export { AmqpAnnotatedMessage } from "./amqpAnnotatedMessage";
 export { logger } from "./log";
+export * from "./internals";

sdk/core/core-amqp/src/internals.ts

@@ -0,0 +1,7 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import { SasTokenProvider, createSasTokenProvider } from "./auth/tokenProvider";
+import { isSasTokenProvider } from "./util/typeGuards";
+
+export { SasTokenProvider, createSasTokenProvider, isSasTokenProvider };

sdk/core/core-amqp/src/util/typeGuards.ts

@@ -1,6 +1,8 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+import { SasTokenProvider } from "../auth/tokenProvider";
+
 /**
  * Helper TypeGuard that checks if something is defined or not.
  * @param thing - Anything
@@ -47,3 +49,12 @@ export function objectHasProperty<Thing extends unknown, PropertyName extends st
 ): thing is Thing & Record<PropertyName, unknown> {
   return typeof thing === "object" && property in (thing as Record<string, unknown>);
 }
+
+/**
+ * Typeguard that checks if the input is a SasTokenProvider.
+ * @param thing - Any object.
+ * @hidden
+ */
+export function isSasTokenProvider(thing: unknown): thing is SasTokenProvider {
+  return isObjectWithProperties(thing, ["isSasTokenProvider"]) && thing.isSasTokenProvider === true;
+}

sdk/core/core-amqp/test/tokenProvider.spec.ts

@@ -0,0 +1,66 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+import chai from "chai";
+const should = chai.should();
+import { AzureNamedKeyCredential, AzureSASCredential } from "@azure/core-auth";
+import { createSasTokenProvider } from "../src/index";
+
+describe("SasTokenProvider", function(): void {
+  describe("createSasTokenProvider", () => {
+    it("should work as expected with AzureNamedKeyCredential", async function(): Promise<void> {
+      const keyName = "myKeyName";
+      const key = "importantValue";
+      const tokenProvider = createSasTokenProvider(new AzureNamedKeyCredential(keyName, key));
+      const now = Math.floor(Date.now() / 1000) + 3600;
+      const tokenInfo = tokenProvider.getToken("myaudience");
+      tokenInfo.token.should.match(
+        /SharedAccessSignature sr=myaudience&sig=(.*)&se=\d{10}&skn=myKeyName/g
+      );
+      tokenInfo.expiresOnTimestamp.should.equal(now);
+    });
+
+    it("should work as expected with `shareAccessKeyName` and `sharedAccessKey`", async function(): Promise<
+      void
+    > {
+      // This is how createSasTokenProvider will be called if SAK params are passed through a connection string.
+      const tokenProvider = createSasTokenProvider({
+        sharedAccessKeyName: "sakName",
+        sharedAccessKey: "sak"
+      });
+      const now = Math.floor(Date.now() / 1000) + 3600;
+      const tokenInfo = tokenProvider.getToken("sb://hostname.servicebus.windows.net/");
+      tokenInfo.token.should.match(
+        /SharedAccessSignature sr=sb%3A%2F%2Fhostname.servicebus.windows.net%2F&sig=(.*)&se=\d{10}&skn=sakName/g
+      );
+      tokenInfo.expiresOnTimestamp.should.equal(now);
+    });
+  });
+
+  it("should work as expected with AzureSASCredential", async function(): Promise<void> {
+    const sasTokenProvider = createSasTokenProvider(
+      new AzureSASCredential("SharedAccessSignature se=<blah>")
+    );
+    const accessToken = sasTokenProvider.getToken("audience isn't used");
+
+    should.equal(
+      accessToken.token,
+      "SharedAccessSignature se=<blah>",
+      "SAS URI we were constructed with should just be returned verbatim without interpretation (and the audience is ignored)"
+    );
+
+    should.equal(
+      accessToken.expiresOnTimestamp,
+      0,
+      "SAS URI always returns 0 for expiry (ignoring what's in the SAS token)"
+    );
+  });
+
+  it("should work as expected with `sharedAccessSignature`", async function(): Promise<void> {
+    // This is how createSasTokenProvider will be called if the shared access signature is passed through a connection string.
+    const tokenProvider = createSasTokenProvider({ sharedAccessSignature: "<blah>" });
+    const tokenInfo = tokenProvider.getToken("sb://hostname.servicebus.windows.net/");
+    tokenInfo.token.should.match(/<blah>/g);
+    tokenInfo.expiresOnTimestamp.should.equal(0);
+  });
+});

sdk/core/core-auth/CHANGELOG.md

@@ -3,7 +3,7 @@
 ## 1.3.0 (Unreleased)
 
 - Adds the `AzureNamedKeyCredential` class which supports credential rotation and a corresponding `NamedKeyCredential` interface to support the use of static string-based names and keys in Azure clients.
-- Adds the `isNamedKeyCredential` and `isSASCredential` typeguard functions.
+- Adds the `isNamedKeyCredential` and `isSASCredential` typeguard functions similar to the existing `isTokenCredential`.
 
 ## 1.2.0 (2021-02-08)
 

sdk/eventhub/event-hubs/CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## 5.5.0 (Unreleased)
 
+- Allows passing `NamedKeyCredential` and `SASCredential` as the credential type to `EventHubConsumerClient` and `EventHubProducerClient`.
+  These credential types support rotation via their `update` methods and are an alternative to using the `SharedAccessKeyName/SharedAccessKey` or `SharedAccessSignature` properties in a connection string.
+
 - Updates the methods on the `CheckpointStore` interface to accept
   an optional `options` parameter that can be used to pass in an
   `abortSignal` and `tracingOptions`.

sdk/eventhub/event-hubs/review/event-hubs.api.md

@@ -6,9 +6,11 @@
 
 import { AbortSignalLike } from '@azure/abort-controller';
 import { MessagingError } from '@azure/core-amqp';
+import { NamedKeyCredential } from '@azure/core-auth';
 import { OperationTracingOptions } from '@azure/core-tracing';
 import { RetryMode } from '@azure/core-amqp';
 import { RetryOptions } from '@azure/core-amqp';
+import { SASCredential } from '@azure/core-auth';
 import { Span } from '@opentelemetry/api';
 import { SpanContext } from '@opentelemetry/api';
 import { TokenCredential } from '@azure/core-auth';
@@ -97,8 +99,8 @@ export class EventHubConsumerClient {
     constructor(consumerGroup: string, connectionString: string, checkpointStore: CheckpointStore, options?: EventHubConsumerClientOptions);
     constructor(consumerGroup: string, connectionString: string, eventHubName: string, options?: EventHubConsumerClientOptions);
     constructor(consumerGroup: string, connectionString: string, eventHubName: string, checkpointStore: CheckpointStore, options?: EventHubConsumerClientOptions);
-    constructor(consumerGroup: string, fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential, options?: EventHubConsumerClientOptions);
-    constructor(consumerGroup: string, fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential, checkpointStore: CheckpointStore, options?: EventHubConsumerClientOptions);
+    constructor(consumerGroup: string, fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential | NamedKeyCredential | SASCredential, options?: EventHubConsumerClientOptions);
+    constructor(consumerGroup: string, fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential | NamedKeyCredential | SASCredential, checkpointStore: CheckpointStore, options?: EventHubConsumerClientOptions);
     close(): Promise<void>;
     static defaultConsumerGroupName: string;
     get eventHubName(): string;
@@ -119,7 +121,7 @@ export interface EventHubConsumerClientOptions extends EventHubClientOptions {
 export class EventHubProducerClient {
     constructor(connectionString: string, options?: EventHubClientOptions);
     constructor(connectionString: string, eventHubName: string, options?: EventHubClientOptions);
-    constructor(fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential, options?: EventHubClientOptions);
+    constructor(fullyQualifiedNamespace: string, eventHubName: string, credential: TokenCredential | NamedKeyCredential | SASCredential, options?: EventHubClientOptions);
     close(): Promise<void>;
     createBatch(options?: CreateBatchOptions): Promise<EventDataBatch>;
     get eventHubName(): string;