Streaming signing

src/packet/packetlist.js

@@ -89,10 +89,7 @@ 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.');
-      }
+    if (util.isStream(packetbytes) && packetParser.supportsStreaming(this[i].tag)) {
       let buffer = [];
       let bufferLength = 0;
       const minLength = 512;
@@ -110,7 +107,14 @@ List.prototype.write = function () {
         }
       }, () => util.concat([packetParser.writeSimpleLength(bufferLength)].concat(buffer))));
     } else {
-      arr.push(packetParser.writeHeader(this[i].tag, packetbytes.length));
+      if (util.isStream(packetbytes)) {
+        let length = 0;
+        arr.push(stream.transform(stream.clone(packetbytes), value => {
+          length += value.length;
+        }, () => packetParser.writeHeader(this[i].tag, length)));
+      } else {
+        arr.push(packetParser.writeHeader(this[i].tag, packetbytes.length));
+      }
       arr.push(packetbytes);
     }
   }

src/packet/signature.js

@@ -239,13 +239,13 @@ Signature.prototype.sign = async function (key, data) {
   }
 
   const toHash = this.toHash(data);
-  const hash = await stream.readToEnd(this.hash(data, toHash));
+  const hash = this.hash(data, toHash);
 
-  this.signedHashValue = hash.subarray(0, 2);
+  this.signedHashValue = stream.slice(stream.clone(hash), 0, 2);
 
-  this.signature = await crypto.signature.sign(
-    publicKeyAlgorithm, hashAlgorithm, key.params, toHash, hash
-  );
+  this.signature = stream.fromAsync(async () => crypto.signature.sign(
+    publicKeyAlgorithm, hashAlgorithm, key.params, toHash, await stream.readToEnd(hash)
+  ));
   return true;
 };
 

test/general/streaming.js

@@ -199,6 +199,51 @@ describe('Streaming', function() {
     }
   });
 
+  it('Encrypt and decrypt larger message roundtrip using public keys (unsafe_stream=true)', async function() {
+    let unsafe_streamValue = openpgp.config.unsafe_stream;
+    openpgp.config.unsafe_stream = true;
+    try {
+      const pubKey = (await openpgp.key.readArmored(pub_key)).keys[0];
+      const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
+      await privKey.decrypt(passphrase);
+
+      let plaintext = [];
+      let i = 0;
+      const data = new ReadableStream({
+        async pull(controller) {
+          await new Promise(setTimeout);
+          if (i++ < 10) {
+            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+            controller.enqueue(randomBytes);
+            plaintext.push(randomBytes);
+          } else {
+            controller.close();
+          }
+        }
+      });
+      const encrypted = await openpgp.encrypt({
+        data,
+        publicKeys: pubKey,
+        privateKeys: privKey
+      });
+
+      const msgAsciiArmored = encrypted.data;
+      const message = await openpgp.message.readArmored(msgAsciiArmored);
+      const decrypted = await openpgp.decrypt({
+        publicKeys: pubKey,
+        privateKeys: privKey,
+        message,
+        format: 'binary'
+      });
+      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).to.deep.equal(plaintext[0]);
+      if (i > 10) throw new Error('Data did not arrive early.');
+      expect(await openpgp.stream.readToEnd(decrypted.data)).to.deep.equal(util.concatUint8Array(plaintext));
+    } finally {
+      openpgp.config.unsafe_stream = unsafe_streamValue;
+    }
+  });
+
   it('Detect MDC modifications (unsafe_stream=true)', async function() {
     let unsafe_streamValue = openpgp.config.unsafe_stream;
     openpgp.config.unsafe_stream = true;