chore: Harden MJPEG frame parsing and stream startup

lib/commands/helpers/mjpeg.ts

@@ -18,7 +18,7 @@ const CONTENT_LENGTH_RE = /Content-Length:\s*(\d+)/i;
  * Extracts individual JPEG frames out of a multipart MJPEG-over-HTTP byte stream by
  * scanning for JPEG start/end-of-image markers and the multipart `Content-Length` header.
  */
-class MjpegFrameParser extends Transform {
+export class MjpegFrameParser extends Transform {
   private buffer: Buffer | null = null;
   private expectedLength = 0;
   private bytesWritten = 0;
@@ -31,7 +31,7 @@ class MjpegFrameParser extends Transform {
     const lengthMatch = CONTENT_LENGTH_RE.exec(chunk.toString('latin1'));
 
     if (this.buffer && (this.isReading || startIdx > -1)) {
-      this.appendChunk(chunk, startIdx, endIdx);
+      this.appendChunk(chunk, endIdx);
     }
     if (lengthMatch) {
       this.startFrame(Number(lengthMatch[1]), chunk, startIdx, endIdx);
@@ -62,15 +62,19 @@ class MjpegFrameParser extends Transform {
     }
   }
 
-  private appendChunk(chunk: Buffer, start: number, end: number): void {
+  private appendChunk(chunk: Buffer, end: number): void {
     if (!this.buffer) {
       return;
     }
-    const copyStart = start > -1 ? start : 0;
+    // We are continuing a frame that is already open, so the whole chunk belongs to it
+    // and must be copied starting at offset 0. A JPEG SOI marker found anywhere in this
+    // chunk can only belong to the *next* frame (it necessarily comes after our own EOI),
+    // so it must never be used as the copy start here, or the current frame's tail bytes
+    // between offset 0 and that marker would be skipped, truncating it.
     const copyEnd = end > -1 ? end + JPEG_EOI.length : chunk.length;
     // Buffer.copy() silently truncates if the destination has less room than requested,
     // so bytesWritten must track what was actually copied, not the requested range size.
-    this.bytesWritten += chunk.copy(this.buffer, this.bytesWritten, copyStart, copyEnd);
+    this.bytesWritten += chunk.copy(this.buffer, this.bytesWritten, 0, copyEnd);
 
     if (end > -1 || this.bytesWritten === this.expectedLength) {
       this.emitFrame();
@@ -82,8 +86,17 @@ class MjpegFrameParser extends Transform {
   private emitFrame(): void {
     this.isReading = false;
     if (this.buffer) {
-      this.push(this.buffer);
+      // Only push what was actually copied: the buffer is allocated to `expectedLength`
+      // up front, so pushing it as-is would leak trailing zero bytes whenever fewer
+      // bytes were actually written (e.g. a mismatched/truncated Content-Length).
+      this.push(this.buffer.subarray(0, this.bytesWritten));
     }
+    // Clear the frame state so an unrelated later chunk (e.g. one with a JPEG SOI
+    // marker but no Content-Length header) cannot be appended onto a buffer that
+    // has already been pushed downstream.
+    this.buffer = null;
+    this.expectedLength = 0;
+    this.bytesWritten = 0;
   }
 }
 
@@ -213,6 +226,10 @@ export class MJpegStream extends Writable {
     const onClose = () => {
       log.debug(`The connection to MJPEG server at ${url} has been closed`);
       this.lastChunk = null;
+      // No-op if start() has already resolved; only rejects a still-pending start().
+      this.registerStartFailure?.(
+        new Error(`The connection to the MJPEG stream at ${url} has been closed before any frame was received`),
+      );
     };
 
     let timeoutId: NodeJS.Timeout | undefined;
@@ -233,6 +250,10 @@ export class MJpegStream extends Writable {
 
     try {
       await startPromise;
+    } catch (err) {
+      // Do not leak the underlying HTTP connection/pipes if we never reached a usable state.
+      this.stop();
+      throw err;
     } finally {
       clearTimeout(timeoutId);
     }

test/unit/commands/helpers/mjpeg.spec.ts

@@ -7,7 +7,12 @@ import sharp from 'sharp';
 import {createSandbox} from 'sinon';
 import type sinon from 'sinon';
 
-import {allocateMjpegServerPort, handleMjpegOptions, MJpegStream} from '../../../../lib/commands/helpers/mjpeg.js';
+import {
+  allocateMjpegServerPort,
+  handleMjpegOptions,
+  MjpegFrameParser,
+  MJpegStream,
+} from '../../../../lib/commands/helpers/mjpeg.js';
 import type {XCUITestDriver} from '../../../../lib/driver.js';
 import {UNIT_LONG_TIMEOUT_MS} from '../../helpers.js';
 
@@ -120,6 +125,83 @@ describe('mjpeg helpers', function () {
       // The server flushes headers immediately, so axios resolves well within the deadline;
       // the rejection comes from MJpegStream's own "no frame yet" guard.
       await expect(stream.start(300)).to.be.rejectedWith(/never sent any images/);
+      // start() must not leak the underlying connection/pipes after failing.
+      expect((stream as any).responseStream).to.be.null;
+    });
+
+    it('should reject quickly if the connection closes before any frame arrives', async function () {
+      const closingServer = http.createServer((_req, res) => {
+        res.writeHead(200, {'Content-Type': 'multipart/x-mixed-replace; boundary=frame'});
+        res.end();
+      });
+      await new Promise<void>((resolve) => closingServer.listen(0, '127.0.0.1', resolve));
+      try {
+        const address = closingServer.address();
+        const port = typeof address === 'object' && address ? address.port : 0;
+        stream = new MJpegStream(`http://127.0.0.1:${port}`);
+        const startedAt = Date.now();
+        // The server timeout is generous; the rejection must come from the close handler,
+        // long before the timeout would otherwise fire.
+        await expect(stream.start(20000)).to.be.rejectedWith(/has been closed/);
+        expect(Date.now() - startedAt).to.be.lessThan(5000);
+        expect((stream as any).responseStream).to.be.null;
+      } finally {
+        await new Promise<void>((resolve) => closingServer.close(() => resolve()));
+      }
+    });
+  });
+
+  describe('MjpegFrameParser', function () {
+    let parser: MjpegFrameParser;
+    let frames: Buffer[];
+
+    beforeEach(function () {
+      parser = new MjpegFrameParser();
+      frames = [];
+      parser.on('data', (frame: Buffer) => frames.push(frame));
+    });
+
+    it('should not leak trailing zero bytes when Content-Length overstates the actual frame size', function (_t, done) {
+      // Declares a 10-byte frame, but the actual SOI..EOI span is only 4 bytes.
+      const chunk = Buffer.concat([Buffer.from('Content-Length: 10\r\n\r\n'), Buffer.from([0xff, 0xd8, 0xff, 0xd9])]);
+      parser.write(chunk, () => {
+        expect(frames).to.have.lengthOf(1);
+        expect(frames[0]).to.deep.equal(Buffer.from([0xff, 0xd8, 0xff, 0xd9]));
+        done();
+      });
+    });
+
+    it('should not corrupt an already-emitted frame with a later stray SOI chunk lacking Content-Length', function (_t, done) {
+      const firstFrame = Buffer.from([0xff, 0xd8, 0xff, 0xd9]);
+      const chunk1 = Buffer.concat([Buffer.from('Content-Length: 10\r\n\r\n'), firstFrame]);
+      parser.write(chunk1, () => {
+        expect(frames).to.have.lengthOf(1);
+        const emitted = frames[0];
+        // No Content-Length header here on purpose: this must not be appended onto
+        // the buffer that was already pushed downstream.
+        const strayChunk = Buffer.from([0xff, 0xd8, 0x01, 0x02, 0x03]);
+        parser.write(strayChunk, () => {
+          expect(frames).to.have.lengthOf(1);
+          expect(emitted).to.deep.equal(firstFrame);
+          done();
+        });
+      });
+    });
+
+    it('should not truncate the current frame when a later chunk also contains a stray SOI marker after the EOI', function (_t, done) {
+      // Frame is split across two writes: the first only carries the SOI (2 of the 4
+      // declared bytes), the second completes it with the EOI, followed by unrelated
+      // trailing bytes that happen to look like another frame's SOI (no Content-Length
+      // alongside them, so they must not be parsed as a new frame).
+      const chunk1 = Buffer.concat([Buffer.from('Content-Length: 4\r\n\r\n'), Buffer.from([0xff, 0xd8])]);
+      const chunk2 = Buffer.from([0xff, 0xd9, 0xff, 0xd8, 0x01]);
+      parser.write(chunk1, () => {
+        parser.write(chunk2, () => {
+          expect(frames).to.have.lengthOf(1);
+          expect(frames[0]).to.deep.equal(Buffer.from([0xff, 0xd8, 0xff, 0xd9]));
+          done();
+        });
+      });
     });
   });