feat(core): add S3 wire checksum download stream

.changeset/stale-rockets-heal.md

@@ -0,0 +1,5 @@
+---
+"@smithy/core": minor
+---
+
+Add createWireChecksumStream to validate S3 wire checksums delivered in the aws-chunked response trailer.

packages/core/src/submodules/serde/index.browser.ts

@@ -92,6 +92,11 @@ export const Hash = no;
 export class Uint8ArrayBlobAdapter extends bindUint8ArrayBlobAdapter(toUtf8, fromUtf8, toBase64, fromBase64) {}
 export { ChecksumStream, type ChecksumStreamInit } from "./util-stream/checksum/ChecksumStream.browser";
 export { createChecksumStream } from "./util-stream/checksum/createChecksumStream.browser";
+export {
+  WireChecksumStream,
+  type WireChecksumStreamInit,
+} from "./util-stream/checksum/wire/WireChecksumStream.browser";
+export { createWireChecksumStream } from "./util-stream/checksum/wire/createWireChecksumStream.browser";
 export { createBufferedReadable } from "./util-stream/createBufferedReadable.browser";
 export { getAwsChunkedEncodingStream } from "./util-stream/getAwsChunkedEncodingStream.browser";
 export { headStream } from "./util-stream/headStream.browser";

packages/core/src/submodules/serde/index.native.ts

@@ -94,6 +94,11 @@ export const Hash = no;
 export class Uint8ArrayBlobAdapter extends bindUint8ArrayBlobAdapter(toUtf8, fromUtf8, toBase64, fromBase64) {}
 export { ChecksumStream, type ChecksumStreamInit } from "./util-stream/checksum/ChecksumStream.browser";
 export { createChecksumStream } from "./util-stream/checksum/createChecksumStream.browser";
+export {
+  WireChecksumStream,
+  type WireChecksumStreamInit,
+} from "./util-stream/checksum/wire/WireChecksumStream.browser";
+export { createWireChecksumStream } from "./util-stream/checksum/wire/createWireChecksumStream.browser";
 export { createBufferedReadable } from "./util-stream/createBufferedReadable.browser";
 export { getAwsChunkedEncodingStream } from "./util-stream/getAwsChunkedEncodingStream.browser";
 export { headStream } from "./util-stream/headStream.browser";

packages/core/src/submodules/serde/index.ts

@@ -90,6 +90,8 @@ export { Hash } from "./hash-node/hash-node";
 export class Uint8ArrayBlobAdapter extends bindUint8ArrayBlobAdapter(toUtf8, fromUtf8, toBase64, fromBase64) {}
 export { ChecksumStream, type ChecksumStreamInit } from "./util-stream/checksum/ChecksumStream";
 export { createChecksumStream } from "./util-stream/checksum/createChecksumStream";
+export { WireChecksumStream, type WireChecksumStreamInit } from "./util-stream/checksum/wire/WireChecksumStream";
+export { createWireChecksumStream } from "./util-stream/checksum/wire/createWireChecksumStream";
 export { createBufferedReadable } from "./util-stream/createBufferedReadable";
 export { getAwsChunkedEncodingStream } from "./util-stream/getAwsChunkedEncodingStream";
 export { headStream } from "./util-stream/headStream";

packages/core/src/submodules/serde/util-stream/checksum/wire/AwsChunkedDecoder.ts

@@ -0,0 +1,213 @@
+/**
+ * Handlers invoked by {@link AwsChunkedDecoder} as it decodes an
+ * `aws-chunked` encoded stream.
+ *
+ * @internal
+ */
+export interface AwsChunkedDecoderHandlers {
+  /**
+   * Called with each segment of decoded data bytes (framing stripped).
+   * May be called multiple times per {@link AwsChunkedDecoder.write}.
+   */
+  onData(chunk: Uint8Array): void;
+  /**
+   * Called once per trailer line found after the terminal chunk.
+   * @param name - the trailer field name (as received, not normalized).
+   * @param value - the trailer field value.
+   */
+  onTrailer(name: string, value: string): void;
+}
+
+const CR = 0x0d; // \r
+const LF = 0x0a; // \n
+
+type State = "CHUNK_HEADER" | "CHUNK_DATA" | "CHUNK_DATA_CRLF" | "TRAILER" | "DONE";
+
+/**
+ * Incremental decoder for `aws-chunked` encoded response bodies.
+ *
+ * Data is fed in arbitrarily sized segments via {@link write}. Decoded data
+ * bytes are surfaced through {@link AwsChunkedDecoderHandlers.onData} as soon
+ * as they are available - the decoder never buffers the full payload. Trailer
+ * fields appended after the terminal chunk are surfaced through
+ * {@link AwsChunkedDecoderHandlers.onTrailer}.
+ *
+ * Wire format (see the S3 Wire Checksum spec):
+ * ```
+ * <HEX-SIZE>[;chunk-ext]\r\n<data bytes>\r\n   (repeated)
+ * 0[;chunk-ext]\r\n                            (terminal chunk, size 0)
+ * <trailer-name>: <value>\r\n                  (zero or more trailers)
+ * \r\n                                         (end of trailers)
+ * ```
+ *
+ * The chunk extension (e.g. `;chunk-signature=UNSIGNED-PAYLOAD`) is parsed and
+ * ignored - no assumption is made about its value, for forward compatibility.
+ *
+ * @internal
+ */
+export class AwsChunkedDecoder {
+  private state: State = "CHUNK_HEADER";
+  /**
+   * Accumulates bytes of the current control line (chunk header or trailer).
+   * Data bytes are never accumulated here - they are emitted directly.
+   */
+  private lineBuffer: number[] = [];
+  /**
+   * Remaining number of data bytes to read for the current chunk.
+   */
+  private chunkRemaining = 0;
+  /**
+   * Remaining number of CRLF bytes to consume after a chunk's data.
+   */
+  private crlfRemaining = 0;
+  /**
+   * Total number of decoded data bytes emitted so far.
+   */
+  private decodedByteCount = 0;
+
+  public constructor(private readonly handlers: AwsChunkedDecoderHandlers) {}
+
+  /**
+   * The number of decoded data bytes emitted so far. After the stream is fully
+   * decoded this equals the `x-amz-decoded-content-length`.
+   */
+  public get bytesDecoded(): number {
+    return this.decodedByteCount;
+  }
+
+  /**
+   * Feed a segment of encoded bytes into the decoder.
+   * @param input - a slice of the aws-chunked encoded stream.
+   */
+  public write(input: Uint8Array): void {
+    let offset = 0;
+    const length = input.length;
+
+    while (offset < length) {
+      switch (this.state) {
+        case "CHUNK_HEADER": {
+          while (offset < length) {
+            const byte = input[offset++];
+            if (byte === LF) {
+              this.parseChunkHeader();
+              break;
+            } else if (byte !== CR) {
+              this.lineBuffer.push(byte);
+            }
+          }
+          break;
+        }
+        case "CHUNK_DATA": {
+          const take = Math.min(this.chunkRemaining, length - offset);
+          if (take > 0) {
+            const data = input.subarray(offset, offset + take);
+            this.decodedByteCount += take;
+            this.chunkRemaining -= take;
+            offset += take;
+            this.handlers.onData(data);
+          }
+          if (this.chunkRemaining === 0) {
+            this.state = "CHUNK_DATA_CRLF";
+            this.crlfRemaining = 2;
+          }
+          break;
+        }
+        case "CHUNK_DATA_CRLF": {
+          while (offset < length && this.crlfRemaining > 0) {
+            const byte = input[offset++];
+            if (byte !== CR && byte !== LF) {
+              throw new Error("@smithy/util-stream: aws-chunked decode error - expected CRLF after chunk data.");
+            }
+            this.crlfRemaining--;
+          }
+          if (this.crlfRemaining === 0) {
+            this.state = "CHUNK_HEADER";
+          }
+          break;
+        }
+        case "TRAILER": {
+          while (offset < length) {
+            const byte = input[offset++];
+            if (byte === LF) {
+              if (this.parseTrailerLine()) {
+                this.state = "DONE";
+              }
+              break;
+            } else if (byte !== CR) {
+              this.lineBuffer.push(byte);
+            }
+          }
+          break;
+        }
+        case "DONE": {
+          // Ignore any bytes received after the trailer terminator.
+          offset = length;
+          break;
+        }
+      }
+    }
+  }
+
+  /**
+   * Signal that the upstream source has ended. Throws if the stream ended
+   * before the terminal chunk and trailers were fully received.
+   */
+  public end(): void {
+    if (this.state !== "DONE") {
+      throw new Error(
+        "@smithy/util-stream: aws-chunked decode error - stream ended before the terminal chunk and trailers were received."
+      );
+    }
+  }
+
+  /**
+   * Parse the accumulated chunk header line and transition state.
+   */
+  private parseChunkHeader(): void {
+    const line = this.takeLine();
+    // Strip any chunk extension (e.g. ";chunk-signature=UNSIGNED-PAYLOAD").
+    const semicolon = line.indexOf(";");
+    const hex = (semicolon === -1 ? line : line.slice(0, semicolon)).trim();
+
+    if (hex.length === 0 || !/^[0-9a-fA-F]+$/.test(hex)) {
+      throw new Error(`@smithy/util-stream: aws-chunked decode error - invalid chunk size "${hex}".`);
+    }
+
+    const size = parseInt(hex, 16);
+    if (size === 0) {
+      this.state = "TRAILER";
+    } else {
+      this.chunkRemaining = size;
+      this.state = "CHUNK_DATA";
+    }
+  }
+
+  /**
+   * Parse the accumulated trailer line.
+   * @returns true when the trailer section is terminated (empty line).
+   */
+  private parseTrailerLine(): boolean {
+    const line = this.takeLine();
+    if (line.length === 0) {
+      return true;
+    }
+    const separator = line.indexOf(":");
+    if (separator === -1) {
+      throw new Error(`@smithy/util-stream: aws-chunked decode error - malformed trailer "${line}".`);
+    }
+    const name = line.slice(0, separator).trim();
+    const value = line.slice(separator + 1).trim();
+    this.handlers.onTrailer(name, value);
+    return false;
+  }
+
+  /**
+   * Consume and return the accumulated control line as a string, resetting the
+   * line buffer.
+   */
+  private takeLine(): string {
+    const line = String.fromCharCode(...this.lineBuffer);
+    this.lineBuffer = [];
+    return line;
+  }
+}

packages/core/src/submodules/serde/util-stream/checksum/wire/WireChecksumStream.browser.ts

@@ -0,0 +1,48 @@
+import type { Checksum, Encoder } from "@smithy/types";
+
+/**
+ * @internal
+ */
+export interface WireChecksumStreamInit {
+  /**
+   * The wire checksum algorithm name (e.g. `crc32`), as reported by the
+   * `x-amz-wire-checksum-algorithm` response header. Used to derive the
+   * trailer field name (`x-amz-wire-checksum-<algorithm>`).
+   */
+  checksumAlgorithm: string;
+
+  /**
+   * The checksum calculator. Updated incrementally over the decoded data bytes.
+   */
+  checksum: Checksum;
+
+  /**
+   * The `aws-chunked` encoded stream carrying the response body. The framing is
+   * decoded before being surfaced, and the checksum value is read from the
+   * trailer.
+   */
+  source: ReadableStream;
+
+  /**
+   * The `x-amz-decoded-content-length` value. When provided, the total decoded
+   * byte count is verified against it. A mismatch indicates a truncated or
+   * corrupt response.
+   */
+  decodedContentLength?: number;
+
+  /**
+   * Optional base 64 encoder if calling from a request context.
+   */
+  base64Encoder?: Encoder;
+}
+
+const ReadableStreamRef = typeof ReadableStream === "function" ? ReadableStream : function (): void {};
+
+/**
+ * This stub exists so that the readable returned by createWireChecksumStream
+ * identifies as "WireChecksumStream" in alignment with the Node.js
+ * implementation.
+ *
+ * @extends ReadableStream
+ */
+export class WireChecksumStream extends (ReadableStreamRef as any) {}