fix: chunked data

Author: panuhorsmalahti

src/__tests__/__snapshots__/stream-impersonator.test.ts.snap

@@ -206,3 +206,35 @@ impersonate-user: johndoe
 
 "
 `;
+
+exports[`StreamImpersonator transfer encoding chunked 1`] = `
+"GET /version HTTP/1.1
+host: 127.0.0.1:53364
+user-agent: node-fetch
+accept: */*
+accept-encoding: gzip, deflate, br
+x-forwarded-for: 127.0.0.1
+authorization: Bearer service-account-token
+impersonate-user: johndoe
+impersonate-group: dev
+impersonate-group: ops
+
+POST /apis/authorization.k8s.io/v1/selfsubjectaccessreviews HTTP/1.1
+host: 127.0.0.1:53364
+user-agent: node-fetch
+transfer-encoding: chunked
+accept: */*
+accept-encoding: gzip, deflate, br
+content-type: application/json
+x-forwarded-for: 127.0.0.1
+authorization: Bearer service-account-token
+impersonate-user: johndoe
+impersonate-group: dev
+impersonate-group: ops
+
+b0
+{"spec":{"resourceAttributes":{"namespace":"kube-system","resource":"*","verb":"create"}},"kind":"SelfSubjectAccessReview","apiVersion":"authorization.k8s.io/v1","metadata":{}}
+0
+
+"
+`;

src/__tests__/stream-impersonator.test.ts

@@ -141,6 +141,43 @@ MwIDAQAB
     expect(destination.buffer.toString()).toContain(`foo\r\nbar`);
   });
 
+  it("transfer encoding chunked", () => {
+    parser.boredServer = boredServer;
+    parser.publicKey = jwtPublicKey;
+
+    const token = jwt.sign({
+      exp: Math.floor(Date.now() / 1000) + (60 * 60),
+      sub: "johndoe",
+      groups: ["dev", "ops"],
+      aud: [boredServer]
+    }, jwtPrivateKey, { algorithm: "RS256" });
+
+    stream.pipe(parser).pipe(destination);
+    stream.write(`GET /version HTTP/1.1\r\n`);
+    stream.write(`Host: 127.0.0.1:53364\r\n`);
+    stream.write(`User-Agent: node-fetch\r\n`);
+    stream.write(`Accept: */*\r\n`);
+    stream.write(`Accept-Encoding: gzip, deflate, br\r\n`);
+    stream.write(`Authorization: Bearer ${token}\r\n`);
+    stream.write(`X-Forwarded-For: 127.0.0.1\r\n\r\n`);
+    stream.write(`POST /apis/authorization.k8s.io/v1/selfsubjectaccessreviews HTTP/1.1\r\n`);
+    stream.write(`Host: 127.0.0.1:53364\r\n`);
+    stream.write(`User-Agent: node-fetch\r\n`);
+    stream.write(`Transfer-Encoding: chunked\r\n`);
+    stream.write(`Accept: */*\r\n`);
+    stream.write(`Accept-Encoding: gzip, deflate, br\r\n`);
+    stream.write(`Authorization: Bearer ${token}\r\n`);
+    stream.write(`Content-Type: application/json\r\n`);
+    stream.write(`X-Forwarded-For: 127.0.0.1\r\n`);
+    stream.write(`\r\n`);
+    stream.write(`b0\r\n`);
+    stream.write(`{"spec":{"resourceAttributes":{"namespace":"kube-system","resource":"*","verb":"create"}},"kind":"SelfSubjectAccessReview","apiVersion":"authorization.k8s.io/v1","metadata":{}}\r\n`);
+    stream.write(`0\r\n`);
+    stream.write(`\r\n`);
+
+    expect(destination.buffer.toString()).toMatchSnapshot();
+  });
+
   it ("handles newline splitted to separate chunks", async () => {
     parser.boredServer = boredServer;
     parser.publicKey = jwtPublicKey;
@@ -287,7 +324,7 @@ MwIDAQAB
     expect(destination.buffer.toString()).toContain("first chunk second chunk");
   });
 
-  it ("handles headers and body in same stream chunk", async () => {
+  it("handles headers and body in same stream chunk", async () => {
     parser.boredServer = boredServer;
     parser.publicKey = jwtPublicKey;
 

src/stream-impersonator.ts

@@ -16,10 +16,13 @@ type Headers = Array<Array<string>>;
 
 type GetSaToken = () => string;
 
+const endOfHeadersMarker = "\r\n\r\n";
+
 export class StreamImpersonator extends Transform {
   public boredServer = "";
   public publicKey = "";
 
+  private headersReceived = false;
   private chunks: Buffer[] = [];
   private httpParser: HTTPParserJS;
   private upgrade = false;
@@ -83,17 +86,20 @@ export class StreamImpersonator extends Transform {
     };
 
     this.httpParser.onBody = (
-      bodyChunk: Buffer,
-      start: number,
-      len: number,
+       
+      _bodyChunk: Buffer,
+       
+      _start: number,
+       
+      _len: number,
     ) => {
-      logger.trace("onBody");
-      this.chunks.push(bodyChunk.subarray(start, start + len));
+      //
     };
 
     this.httpParser.onMessageComplete = () => {
       logger.trace("onMessageComplete");
       this.flushChunks();
+      this.headersReceived = false;
     };
   }
 
@@ -128,14 +134,45 @@ export class StreamImpersonator extends Transform {
       return callback();
     }
 
+    this.chunks.push(chunk);
     this.partialMessage.push(chunk);
 
+    const receivedSoFar = Buffer.concat(this.partialMessage);
+
+    // +endOfHeadersMarker.length to include the \r\n\r\n in the header
+    const headerEndIndex = receivedSoFar.indexOf(endOfHeadersMarker) + endOfHeadersMarker.length;
+
+    // Wait for more data if headers are incomplete and not received yet
+    if (headerEndIndex === -1 && !this.headersReceived) {
+      return callback();
+    }
+
     const handleError = (err: Error) => {
       this.partialMessage = [];
       logger.error("[IMPERSONATOR] Error parsing HTTP data: %s", String(err));
       throw err;
     };
 
+    // Parse headers if not parsed yet
+    if (!this.headersReceived) {
+      this.headersReceived = true;
+
+      // Extract headers
+      const bufferToParse = Buffer.concat(this.partialMessage).subarray(0, headerEndIndex);
+
+      const bytesParsed = this.httpParser.execute(bufferToParse);
+
+      if (bytesParsed instanceof Error) {
+        return handleError(bytesParsed);
+      }
+
+      // Remove parsed headers from the buffer
+      this.partialMessage = removeBytesFromBuffersHead(this.partialMessage, bytesParsed);
+
+      // onHeadersComplete set this.chunks to [], we set the rest of the bytes after the header back
+      this.chunks = [Buffer.concat(this.partialMessage)];
+    }
+
     try {
       const bufferToParse = Buffer.concat(this.partialMessage);
       const bytesParsed = this.httpParser.execute(bufferToParse);