Streaming decrypt non-partial-length-encoded packets

src/packet/packet.js

@@ -15,13 +15,18 @@
 // License along with this library; if not, write to the Free Software
 // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 
+/* eslint-disable callback-return */
+
 /**
  * @fileoverview Functions for reading and writing packets
  * @requires enums
+ * @requires stream
  * @requires util
  * @module packet/packet
  */
 
+import enums from '../enums';
+import stream from '../stream';
 import util from '../util';
 
 export default {
@@ -110,23 +115,39 @@ export default {
     return util.concatUint8Array([new Uint8Array([0x80 | (tag_type << 2) | 2]), util.writeNumber(length, 4)]);
   },
 
+  /**
+   * Whether the packet type supports partial lengths per RFC4880
+   * @param {Integer} tag_type Tag type
+   * @returns {Boolean} String of the header
+   */
+  supportsStreaming: function(tag_type) {
+    return [
+      enums.packet.literal,
+      enums.packet.compressed,
+      enums.packet.symmetricallyEncrypted,
+      enums.packet.symEncryptedIntegrityProtected,
+      enums.packet.symEncryptedAEADProtected
+    ].includes(tag_type);
+  },
+
   /**
    * Generic static Packet Parser function
    *
-   * @param {String} input Input stream as string
+   * @param {Uint8Array} input Input stream as string
-   * @param {integer} position Position to start parsing
+   * @param {Function} callback Function to call with the parsed packet
-   * @param {integer} len Length of the input from position on
+   * @returns {Boolean} Returns false if the stream was empty and parsing is done, and true otherwise.
-   * @returns {Object} Returns a parsed module:packet/packet
    */
-  read: function(reader) {
+  read: async function(input, callback) {
-    return new Promise(async (resolve, reject) => {
+    let reader = stream.getReader(input);
     let controller;
     try {
       const peekedBytes = await reader.peekBytes(2);
+      if (!peekedBytes || !peekedBytes.length) {
+        return false;
+      }
       // some sanity checks
       if (!peekedBytes || peekedBytes.length < 2 || (peekedBytes[0] & 0x80) === 0) {
-          reject(new Error("Error during parsing. This message / key probably does not conform to a valid OpenPGP format."));
+        throw new Error("Error during parsing. This message / key probably does not conform to a valid OpenPGP format.");
-          return;
       }
       const headerByte = await reader.readByte();
       let tag = -1;
@@ -148,7 +169,7 @@ export default {
         packet_length_type = headerByte & 0x03; // bit 1-0
       }
 
-        let bodydata = null;
+      let packet = null;
       if (!format) {
         // 4.2.1. Old Format Packet Lengths
         switch (packet_length_type) {
@@ -183,8 +204,12 @@ export default {
             break;
         }
       } else { // 4.2.2. New Format Packet Lengths
+        const streaming = this.supportsStreaming(tag);
+        let wasPartialLength;
+        do {
           // 4.2.2.1. One-Octet Lengths
           const lengthByte = await reader.readByte();
+          wasPartialLength = false;
           if (lengthByte < 192) {
             packet_length = lengthByte;
             // 4.2.2.2. Two-Octet Lengths
@@ -193,69 +218,55 @@ export default {
             // 4.2.2.4. Partial Body Lengths
           } else if (lengthByte > 223 && lengthByte < 255) {
             packet_length = 1 << (lengthByte & 0x1F);
-            bodydata = new ReadableStream({
+            wasPartialLength = true;
+            if (!streaming) {
+              throw new TypeError('This packet type does not support partial lengths.');
+            } else if (!packet) {
+              packet = new ReadableStream({
+                // eslint-disable-next-line no-loop-func
                 async start(_controller) {
                   controller = _controller;
                 }
               });
-            resolve({
+              callback({ tag, packet });
-              tag: tag,
-              packet: bodydata,
-              done: true
-            });
-            controller.enqueue(await reader.readBytes(packet_length));
-            let tmplen;
-            while (true) {
-              const tmplenByte = await reader.readByte();
-              if (tmplenByte < 192) {
-                tmplen = tmplenByte;
-                controller.enqueue(await reader.readBytes(tmplen));
-                break;
-              } else if (tmplenByte >= 192 && tmplenByte < 224) {
-                tmplen = ((tmplenByte - 192) << 8) + (await reader.readByte()) + 192;
-                controller.enqueue(await reader.readBytes(tmplen));
-                break;
-              } else if (tmplenByte > 223 && tmplenByte < 255) {
-                tmplen = 1 << (tmplenByte & 0x1F);
-                controller.enqueue(await reader.readBytes(tmplen));
-              } else {
-                tmplen = (await reader.readByte() << 24) | (await reader.readByte() << 16) | (await reader.readByte() << 8) | await reader.readByte();
-                controller.enqueue(await reader.readBytes(tmplen));
-                break;
-              }
             }
             // 4.2.2.3. Five-Octet Lengths
           } else {
             packet_length = (await reader.readByte() << 24) | (await reader.readByte() << 16) | (await reader.readByte() <<
               8) | await reader.readByte();
           }
+          if (streaming) {
+            if (controller) {
+              controller.enqueue(await reader.readBytes(packet_length));
+            } else {
+              // Send the remainder of the packet to the callback as a stream
+              reader.releaseLock();
+              packet = stream.subarray(stream.clone(input), 0, packet_length);
+              await callback({ tag, packet });
+
+              // Read the entire packet before parsing the next one
+              reader = stream.getReader(input);
+              await reader.readBytes(packet_length);
+            }
+          }
+        } while(wasPartialLength);
       }
 
-        // if there wasn't a partial body length
+      if (!packet) {
-        if (bodydata === null) {
+        packet = await reader.readBytes(packet_length);
-          bodydata = await reader.readBytes(packet_length);
+        await callback({ tag, packet });
-
+      } else if (controller) {
-          const peekedByte = await reader.peekBytes(1);
-          resolve({
-            tag: tag,
-            packet: bodydata,
-            done: !(peekedByte && peekedByte.length)
-          });
-        } else {
-          const { done, value } = await reader.read();
-          if (!done && value.length) {
-            throw new Error('Packets after a packet with partial lengths are not supported');
-          } else {
         controller.close();
       }
-        }
     } catch(e) {
       if (controller) {
         controller.error(e);
       } else {
-          reject(e);
+        throw e;
       }
+    } finally {
+      reader.releaseLock();
     }
-    });
+    return true;
   }
 };

src/packet/packetlist.js

@@ -36,31 +36,40 @@ function List() {
  * @param {Uint8Array} A Uint8Array of bytes.
  */
 List.prototype.read = async function (bytes) {
-  const reader = stream.getReader(bytes);
+  this.stream = new ReadableStream({
-  while (true) {
+    pull: async controller => {
-    const parsed = await packetParser.read(reader);
+      if (!await packetParser.read(bytes, async parsed => {
-
-    let pushed = false;
         try {
           const tag = enums.read(enums.packet, parsed.tag);
           const packet = packets.newPacketFromTag(tag);
-      this.push(packet);
+          packet.packets = new List();
-      pushed = true;
+          packet.fromStream = util.isStream(parsed.packet);
           await packet.read(parsed.packet);
-      if (parsed.done) {
+          controller.enqueue(packet);
-        break;
-      }
         } catch (e) {
           if (!config.tolerant ||
               parsed.tag === enums.packet.symmetricallyEncrypted ||
               parsed.tag === enums.packet.literal ||
               parsed.tag === enums.packet.compressed) {
-        throw e;
+            controller.error(e);
           }
           util.print_debug_error(e);
-      if (pushed) {
-        this.pop(); // drop unsupported packet
         }
+      })) {
+        controller.close();
+      }
+    }
+  });
+
+  // Wait until first few packets have been read
+  const reader = stream.getReader(stream.clone(this.stream));
+  while (true) {
+    const { done, value } = await reader.read();
+    if (!done) {
+      this.push(value);
+    }
+    if (done || value.fromStream) {
+      break;
     }
   }
 };
@@ -76,6 +85,9 @@ List.prototype.write = function () {
   for (let i = 0; i < this.length; i++) {
     const packetbytes = this[i].write();
     if (util.isStream(packetbytes)) {
+      if (!packetParser.supportsStreaming(this[i].tag)) {
+        throw new Error('This packet type does not support partial lengths.');
+      }
       let buffer = [];
       let bufferLength = 0;
       const minLength = 512;

src/stream.js

@@ -52,7 +52,9 @@ function transform(input, process = () => undefined, finish = () => undefined) {
 
 function tee(input) {
   if (util.isStream(input)) {
-    return input.tee();
+    const teed = input.tee();
+    teed[0].externalBuffer = teed[1].externalBuffer = input.externalBuffer;
+    return teed;
   }
   return [input, input];
 }
@@ -202,9 +204,14 @@ ReadableStream.prototype.tee = function() {
 
 
 function Reader(input) {
+  this.stream = input;
+  if (input.externalBuffer) {
+    this.externalBuffer = input.externalBuffer.slice();
+  }
   if (util.isStream(input)) {
     const reader = input.getReader();
     this._read = reader.read.bind(reader);
+    this._releaseLock = reader.releaseLock.bind(reader);
     return;
   }
   let doneReading = false;
@@ -215,6 +222,7 @@ function Reader(input) {
     doneReading = true;
     return { value: input, done: false };
   };
+  this._releaseLock = () => {};
 }
 
 Reader.prototype.read = async function() {
@@ -225,6 +233,11 @@ Reader.prototype.read = async function() {
   return this._read();
 };
 
+Reader.prototype.releaseLock = function() {
+  this.stream.externalBuffer = this.externalBuffer;
+  this._releaseLock();
+};
+
 Reader.prototype.readLine = async function() {
   let buffer = [];
   let returnVal;