feat(event-stream): update event stream handler to use message signer

Author: AndrewFossAWS

packages/eventstream-handler-node/src/EventSigningStream.spec.ts

@@ -1,5 +1,5 @@
 import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
-import { Message, MessageHeaders } from "@aws-sdk/types";
+import { Message, MessageHeaders, SignedMessage } from "@aws-sdk/types";
 import { fromUtf8, toUtf8 } from "@aws-sdk/util-utf8";
 
 import { EventSigningStream } from "./EventSigningStream";
@@ -13,18 +13,17 @@ describe("EventSigningStream", () => {
 
   it("should sign a eventstream payload properly", (done) => {
     const eventStreamCodec = new EventStreamCodec(toUtf8, fromUtf8);
-    const inputChunks: Array<Uint8Array> = (
-      [
-        {
-          headers: {},
-          body: fromUtf8("foo"),
-        },
-        {
-          headers: {},
-          body: fromUtf8("bar"),
-        },
-      ] as Array<Message>
-    ).map((event) => eventStreamCodec.encode(event));
+    const message1: Message = {
+      headers: {},
+      body: fromUtf8("foo"),
+    };
+    const message2: Message = {
+      headers: {},
+      body: fromUtf8("bar"),
+    };
+    const inputChunks: Array<Uint8Array> = ([message1, message2] as Array<Message>).map((event) =>
+      eventStreamCodec.encode(event)
+    );
     const expected: Array<MessageHeaders> = [
       {
         ":date": { type: "timestamp", value: new Date(1546045446000) },
@@ -41,10 +40,10 @@ describe("EventSigningStream", () => {
         },
       },
     ];
-    const mockEventSigner = jest
+    const mockMessageSigner = jest
       .fn()
-      .mockReturnValueOnce("7369676e617475726531") //'signature1'
-      .mockReturnValueOnce("7369676e617475726532"); //'signature2'
+      .mockReturnValueOnce({ message: message1, signature: "7369676e617475726531" } as SignedMessage) //'signature1'
+      .mockReturnValueOnce({ message: message2, signature: "7369676e617475726532" } as SignedMessage); //'signature2'
     // mock 'new Date()'
     let mockDateCount = 0;
     // eslint-disable-next-line @typescript-eslint/no-unused-vars
@@ -58,7 +57,10 @@ describe("EventSigningStream", () => {
       });
     const signingStream = new EventSigningStream({
       priorSignature: "initial",
-      eventSigner: { sign: mockEventSigner },
+      messageSigner: {
+        sign: mockMessageSigner,
+        signMessage: mockMessageSigner,
+      },
       eventStreamCodec,
     });
     const output: Array<MessageHeaders> = [];
@@ -67,12 +69,12 @@ describe("EventSigningStream", () => {
     });
     signingStream.on("end", () => {
       expect(output).toEqual(expected);
-      expect(mockEventSigner.mock.calls[0][1].priorSignature).toBe("initial");
-      expect(mockEventSigner.mock.calls[0][1].signingDate.getTime()).toBe(
+      expect(mockMessageSigner.mock.calls[0][0].priorSignature).toBe("initial");
+      expect(mockMessageSigner.mock.calls[0][1].signingDate.getTime()).toBe(
         (expected[0][":date"].value as Date).getTime()
       );
-      expect(mockEventSigner.mock.calls[1][1].priorSignature).toBe("7369676e617475726531");
-      expect(mockEventSigner.mock.calls[1][1].signingDate.getTime()).toBe(
+      expect(mockMessageSigner.mock.calls[1][0].priorSignature).toBe("7369676e617475726531");
+      expect(mockMessageSigner.mock.calls[1][1].signingDate.getTime()).toBe(
         (expected[1][":date"].value as Date).getTime()
       );
       done();

packages/eventstream-handler-node/src/EventSigningStream.ts

@@ -1,24 +1,24 @@
 import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
-import { EventSigner, MessageHeaders } from "@aws-sdk/types";
+import { MessageHeaders, MessageSigner } from "@aws-sdk/types";
 import { Transform, TransformCallback, TransformOptions } from "stream";
 
 /**
  * @internal
  */
 export interface EventSigningStreamOptions extends TransformOptions {
   priorSignature: string;
-  eventSigner: EventSigner;
+  messageSigner: MessageSigner;
   eventStreamCodec: EventStreamCodec;
 }
 
 /**
  * @internal
- * 
+ *
  * A transform stream that signs the eventstream
  */
 export class EventSigningStream extends Transform {
   private priorSignature: string;
-  private eventSigner: EventSigner;
+  private messageSigner: MessageSigner;
   private eventStreamCodec: EventStreamCodec;
 
   constructor(options: EventSigningStreamOptions) {
@@ -30,8 +30,8 @@ export class EventSigningStream extends Transform {
     });
 
     this.priorSignature = options.priorSignature;
-    this.eventSigner = options.eventSigner;
     this.eventStreamCodec = options.eventStreamCodec;
+    this.messageSigner = options.messageSigner;
   }
 
   async _transform(chunk: Uint8Array, encoding: string, callback: TransformCallback): Promise<void> {
@@ -40,23 +40,25 @@ export class EventSigningStream extends Transform {
       const dateHeader: MessageHeaders = {
         ":date": { type: "timestamp", value: now },
       };
-      const signature = await this.eventSigner.sign(
+      const signedMessage = await this.messageSigner.sign(
         {
-          payload: chunk,
-          headers: this.eventStreamCodec.formatHeaders(dateHeader),
+          message: {
+            body: chunk,
+            headers: dateHeader,
+          },
+          priorSignature: this.priorSignature,
         },
         {
-          priorSignature: this.priorSignature,
           signingDate: now,
         }
       );
-      this.priorSignature = signature;
+      this.priorSignature = signedMessage.signature;
       const serializedSigned = this.eventStreamCodec.encode({
         headers: {
           ...dateHeader,
           ":chunk-signature": {
             type: "binary",
-            value: getSignatureBinary(signature),
+            value: getSignatureBinary(signedMessage.signature),
           },
         },
         body: chunk,

packages/eventstream-handler-node/src/EventStreamPayloadHandler.spec.ts

@@ -1,5 +1,12 @@
 import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
-import { Decoder, Encoder, EventSigner, FinalizeHandler, FinalizeHandlerArguments, HttpRequest } from "@aws-sdk/types";
+import {
+  Decoder,
+  Encoder,
+  FinalizeHandler,
+  FinalizeHandlerArguments,
+  HttpRequest,
+  MessageSigner,
+} from "@aws-sdk/types";
 import { PassThrough, Readable } from "stream";
 
 import { EventSigningStream } from "./EventSigningStream";
@@ -9,8 +16,9 @@ jest.mock("./EventSigningStream");
 jest.mock("@aws-sdk/eventstream-codec");
 
 describe(EventStreamPayloadHandler.name, () => {
-  const mockSigner: EventSigner = {
+  const mockMessageSigner: MessageSigner = {
     sign: jest.fn(),
+    signMessage: jest.fn(),
   };
   const mockUtf8Decoder: Decoder = jest.fn();
   const mockUtf8encoder: Encoder = jest.fn();
@@ -27,7 +35,7 @@ describe(EventStreamPayloadHandler.name, () => {
 
   it("should throw if request payload is not a stream", () => {
     const handler = new EventStreamPayloadHandler({
-      eventSigner: () => Promise.resolve(mockSigner),
+      messageSigner: () => Promise.resolve(mockMessageSigner),
       utf8Decoder: mockUtf8Decoder,
       utf8Encoder: mockUtf8encoder,
     });
@@ -44,7 +52,7 @@ describe(EventStreamPayloadHandler.name, () => {
     (mockNextHandler as any).mockImplementationOnce(() => Promise.reject(mockError));
 
     const handler = new EventStreamPayloadHandler({
-      eventSigner: () => Promise.resolve(mockSigner),
+      messageSigner: () => Promise.resolve(mockMessageSigner),
       utf8Decoder: mockUtf8Decoder,
       utf8Encoder: mockUtf8encoder,
     });
@@ -74,7 +82,7 @@ describe(EventStreamPayloadHandler.name, () => {
     } as any;
 
     const handler = new EventStreamPayloadHandler({
-      eventSigner: () => Promise.resolve(mockSigner),
+      messageSigner: () => Promise.resolve(mockMessageSigner),
       utf8Decoder: mockUtf8Decoder,
       utf8Encoder: mockUtf8encoder,
     });
@@ -88,7 +96,7 @@ describe(EventStreamPayloadHandler.name, () => {
     expect(EventSigningStream).toHaveBeenCalledWith({
       priorSignature,
       eventStreamCodec: expect.anything(),
-      eventSigner: expect.anything(),
+      messageSigner: expect.anything(),
     });
   });
 
@@ -105,7 +113,7 @@ describe(EventStreamPayloadHandler.name, () => {
     } as any;
 
     const handler = new EventStreamPayloadHandler({
-      eventSigner: () => Promise.resolve(mockSigner),
+      messageSigner: () => Promise.resolve(mockMessageSigner),
       utf8Decoder: mockUtf8Decoder,
       utf8Encoder: mockUtf8encoder,
     });
@@ -119,7 +127,7 @@ describe(EventStreamPayloadHandler.name, () => {
     expect(EventSigningStream).toHaveBeenCalledWith({
       priorSignature,
       eventStreamCodec: expect.anything(),
-      eventSigner: expect.anything(),
+      messageSigner: expect.anything(),
     });
   });
 
@@ -132,7 +140,7 @@ describe(EventStreamPayloadHandler.name, () => {
       headers: { authorization },
     } as any;
     const handler = new EventStreamPayloadHandler({
-      eventSigner: () => Promise.resolve(mockSigner),
+      messageSigner: () => Promise.resolve(mockMessageSigner),
       utf8Decoder: mockUtf8Decoder,
       utf8Encoder: mockUtf8encoder,
     });

packages/eventstream-handler-node/src/EventStreamPayloadHandler.ts

@@ -2,7 +2,6 @@ import { EventStreamCodec } from "@aws-sdk/eventstream-codec";
 import {
   Decoder,
   Encoder,
-  EventSigner,
   EventStreamPayloadHandler as IEventStreamPayloadHandler,
   FinalizeHandler,
   FinalizeHandlerArguments,
@@ -21,7 +20,7 @@ import { EventSigningStream } from "./EventSigningStream";
  * @internal
  */
 export interface EventStreamPayloadHandlerOptions {
-  eventSigner: Provider<EventSigner>;
+  messageSigner: Provider<MessageSigner>;
   utf8Encoder: Encoder;
   utf8Decoder: Decoder;
 }
@@ -36,11 +35,11 @@ export interface EventStreamPayloadHandlerOptions {
  * 4. Sign the payload after payload stream starting to flow.
  */
 export class EventStreamPayloadHandler implements IEventStreamPayloadHandler {
-  private readonly eventSigner: Provider<EventSigner>;
+  private readonly messageSigner: Provider<MessageSigner>;
   private readonly eventStreamCodec: EventStreamCodec;
 
   constructor(options: EventStreamPayloadHandlerOptions) {
-    this.eventSigner = options.eventSigner;
+    this.messageSigner = options.messageSigner;
     this.eventStreamCodec = new EventStreamCodec(options.utf8Encoder, options.utf8Decoder);
   }
 
@@ -79,7 +78,7 @@ export class EventStreamPayloadHandler implements IEventStreamPayloadHandler {
     const signingStream = new EventSigningStream({
       priorSignature,
       eventStreamCodec: this.eventStreamCodec,
-      eventSigner: await this.eventSigner(),
+      messageSigner: await this.messageSigner(),
     });
 
     pipeline(payloadStream, signingStream, request.body, (err: NodeJS.ErrnoException | null) => {

packages/eventstream-handler-node/src/provider.ts

@@ -1,10 +1,17 @@
-import { Decoder, Encoder, EventSigner, EventStreamPayloadHandlerProvider, Provider } from "@aws-sdk/types";
+import {
+  Decoder,
+  Encoder,
+  EventSigner,
+  EventStreamPayloadHandlerProvider,
+  MessageSigner,
+  Provider,
+} from "@aws-sdk/types";
 
 import { EventStreamPayloadHandler } from "./EventStreamPayloadHandler";
 
 /** NodeJS event stream utils provider */
 export const eventStreamPayloadHandlerProvider: EventStreamPayloadHandlerProvider = (options: {
   utf8Encoder: Encoder;
   utf8Decoder: Decoder;
-  eventSigner: Provider<EventSigner>;
+  messageSigner: Provider<MessageSigner>;
 }) => new EventStreamPayloadHandler(options);

packages/middleware-eventstream/src/configuration.ts

@@ -32,9 +32,10 @@ export function resolveEventStreamConfig<T>(
   input: T & PreviouslyResolved & EventStreamInputConfig
 ): T & EventStreamResolvedConfig {
   const eventSigner = input.signer;
+  const messageSigner = input.signer;
   const eventStreamPayloadHandler = input.eventStreamPayloadHandlerProvider({
     ...input,
-    eventSigner,
+    messageSigner,
   });
   return {
     ...input,

packages/types/src/signature.ts

@@ -164,4 +164,5 @@ export interface SignedMessage {
  */
 export interface MessageSigner {
   signMessage(message: SignableMessage, args: SigningArguments): Promise<SignedMessage>;
+  sign(event: SignableMessage, options: SigningArguments): Promise<SignedMessage>;
 }