Replace stream.tee() with stream.clone()

Also some other fixes to pass more tests.

src/encoding/armor.js

@@ -215,13 +215,13 @@ function dearmor(input) {
       let text = [];
       let textDone;
       let controller;
-      let [data, dataClone] = stream.tee(base64.decode(new ReadableStream({
+      let data = base64.decode(new ReadableStream({
         async start(_controller) {
           controller = _controller;
         }
-      })));
+      }));
       let checksum;
-      const checksumVerified = getCheckSum(dataClone);
+      const checksumVerified = getCheckSum(stream.clone(data));
       data = stream.getReader(data).substream(); // Convert to Stream
       data = stream.transform(data, value => value, async () => {
         const checksumVerifiedString = await stream.readToEnd(checksumVerified);
@@ -303,8 +303,7 @@ function armor(messagetype, body, partindex, parttotal, customComment) {
     hash = body.hash;
     body = body.data;
   }
-  let bodyClone;
+  const bodyClone = stream.clone(body);
-  [body, bodyClone] = stream.tee(body);
   const result = [];
   switch (messagetype) {
     case enums.armor.multipart_section:

src/openpgp.js

@@ -324,14 +324,14 @@ export function encrypt({ data, dataType, publicKeys, privateKeys, passwords, se
     return message.encrypt(publicKeys, passwords, sessionKey, wildcard, date, toUserId);
 
   }).then(async encrypted => {
-    let message = encrypted.message;
     if (armor) {
-      message = message.armor();
+      result.data = encrypted.message.armor();
+      if (!util.isStream(data)) {
+        result.data = await stream.readToEnd(result.data);
+      }
+    } else {
+      result.message = encrypted.message;
     }
-    if (util.isStream(message) && !util.isStream(data)) {
-      message = await stream.readToEnd(message);
-    }
-    result[armor ? 'data' : 'message'] = message;
     if (returnSessionKey) {
       result.sessionKey = encrypted.sessionKey;
     }
@@ -420,6 +420,9 @@ export function sign({ data, dataType, privateKeys, armor=true, detached=false,
       message = await message.sign(privateKeys, undefined, date, fromUserId);
       if (armor) {
         result.data = message.armor();
+        if (!util.isStream(data)) {
+          result.data = await stream.readToEnd(result.data);
+        }
       } else {
         result.message = message;
       }

src/packet/literal.js

@@ -66,11 +66,9 @@ function normalize(text) {
  * @returns {String} literal data as text
  */
 Literal.prototype.getText = function() {
-  let text;
   if (this.text === null) {
     let lastChar = '';
-    [this.data, this.text] = stream.tee(this.data);
+    this.text = stream.transform(stream.clone(this.data), value => {
-    this.text = stream.transform(this.text, value => {
       const text = lastChar + util.Uint8Array_to_str(value);
       // decode UTF8 and normalize EOL to \n
       const normalized = normalize(text);
@@ -84,8 +82,7 @@ Literal.prototype.getText = function() {
       return normalized.slice(0, -1);
     }, () => lastChar);
   }
-  [text, this.text] = stream.tee(this.text);
+  return stream.clone(this.text);
-  return text;
 };
 
 /**
@@ -105,15 +102,13 @@ Literal.prototype.setBytes = function(bytes, format) {
  * @returns {Uint8Array} A sequence of bytes
  */
 Literal.prototype.getBytes = function() {
-  if (this.data !== null) {
+  if (this.data === null) {
-    return this.data;
+    // normalize EOL to \r\n
+    const text = util.canonicalizeEOL(this.text);
+    // encode UTF8
+    this.data = util.str_to_Uint8Array(util.encode_utf8(text));
   }
-
+  return stream.clone(this.data);
-  // normalize EOL to \r\n
-  const text = util.canonicalizeEOL(this.text);
-  // encode UTF8
-  this.data = util.str_to_Uint8Array(util.encode_utf8(text));
-  return this.data;
 };
 
 

src/packet/sym_encrypted_integrity_protected.js

@@ -94,13 +94,14 @@ SymEncryptedIntegrityProtected.prototype.encrypt = async function (sessionKeyAlg
   const prefix = util.concat([prefixrandom, repeat]);
   const mdc = new Uint8Array([0xD3, 0x14]); // modification detection code packet
 
-  let [tohash, tohashClone] = stream.tee(util.concat([bytes, mdc]));
+  let tohash = util.concat([bytes, mdc]);
-  const hash = crypto.hash.sha1(util.concat([prefix, tohashClone]));
+  const hash = crypto.hash.sha1(util.concat([prefix, stream.clone(tohash)]));
   tohash = util.concat([tohash, hash]);
 
   if (sessionKeyAlgorithm.substr(0, 3) === 'aes') { // AES optimizations. Native code for node, asmCrypto for browser.
     this.encrypted = aesEncrypt(sessionKeyAlgorithm, util.concat([prefix, tohash]), key);
   } else {
+    tohash = await stream.readToEnd(tohash);
     this.encrypted = crypto.cfb.encrypt(prefixrandom, sessionKeyAlgorithm, tohash, key, false);
     this.encrypted = stream.subarray(this.encrypted, 0, prefix.length + tohash.length);
   }
@@ -115,29 +116,28 @@ SymEncryptedIntegrityProtected.prototype.encrypt = async function (sessionKeyAlg
  * @async
  */
 SymEncryptedIntegrityProtected.prototype.decrypt = async function (sessionKeyAlgorithm, key) {
-  const [encrypted, encryptedClone] = stream.tee(this.encrypted);
+  const encrypted = stream.clone(this.encrypted);
+  const encryptedClone = stream.clone(encrypted);
   let decrypted;
   if (sessionKeyAlgorithm.substr(0, 3) === 'aes') { // AES optimizations. Native code for node, asmCrypto for browser.
     decrypted = aesDecrypt(sessionKeyAlgorithm, encrypted, key);
   } else {
-    decrypted = crypto.cfb.decrypt(sessionKeyAlgorithm, key, encrypted, false);
+    decrypted = crypto.cfb.decrypt(sessionKeyAlgorithm, key, await stream.readToEnd(encrypted), false);
   }
 
-  let decryptedClone;
-  [decrypted, decryptedClone] = stream.tee(decrypted);
   // there must be a modification detection code packet as the
   // last packet and everything gets hashed except the hash itself
   const encryptedPrefix = await stream.readToEnd(stream.subarray(encryptedClone, 0, crypto.cipher[sessionKeyAlgorithm].blockSize + 2));
   const prefix = crypto.cfb.mdc(sessionKeyAlgorithm, key, encryptedPrefix);
-  let [bytes, bytesClone] = stream.tee(stream.subarray(decrypted, 0, -20));
+  const bytes = stream.subarray(stream.clone(decrypted), 0, -20);
-  const tohash = util.concat([prefix, bytes]);
+  const tohash = util.concat([prefix, stream.clone(bytes)]);
   this.hash = util.Uint8Array_to_str(await stream.readToEnd(crypto.hash.sha1(tohash)));
-  const mdc = util.Uint8Array_to_str(await stream.readToEnd(stream.subarray(decryptedClone, -20)));
+  const mdc = util.Uint8Array_to_str(await stream.readToEnd(stream.subarray(decrypted, -20)));
 
   if (this.hash !== mdc) {
     throw new Error('Modification detected.');
   } else {
-    await this.packets.read(stream.subarray(bytesClone, 0, -2));
+    await this.packets.read(stream.subarray(bytes, 0, -2));
   }
 
   return true;

src/stream.js

@@ -51,6 +51,16 @@ function tee(input) {
   return [input, input];
 }
 
+function clone(input) {
+  if (util.isStream(input)) {
+    const teed = tee(input);
+    input.getReader = teed[0].getReader.bind(teed[0]);
+    input.tee = teed[0].tee.bind(teed[0]);
+    return teed[1];
+  }
+  return input;
+}
+
 function subarray(input, begin=0, end=Infinity) {
   if (util.isStream(input)) {
     if (begin >= 0 && end >= 0) {
@@ -90,7 +100,7 @@ async function readToEnd(input, join) {
 }
 
 
-export default { concat, getReader, transform, tee, subarray, readToEnd };
+export default { concat, getReader, transform, clone, subarray, readToEnd };
 
 
 /*const readerAcquiredMap = new Map();
@@ -103,6 +113,16 @@ ReadableStream.prototype.getReader = function() {
     readerAcquiredMap.set(this, new Error('Reader for this ReadableStream already acquired here.'));
   }
   return _getReader.apply(this, arguments);
+};
+
+const _tee = ReadableStream.prototype.tee;
+ReadableStream.prototype.tee = function() {
+  if (readerAcquiredMap.has(this)) {
+    console.error(readerAcquiredMap.get(this));
+  } else {
+    readerAcquiredMap.set(this, new Error('Reader for this ReadableStream already acquired here.'));
+  }
+  return _tee.apply(this, arguments);
 };*/
 
 

src/util.js

@@ -432,16 +432,14 @@ export default {
     }
   },
 
-  print_entire_stream: function (str, stream, fn = result => result) {
+  print_entire_stream: function (str, input, fn = result => result) {
-    const teed = stream.tee();
+    stream.readToEnd(stream.clone(input)).then(result => {
-    stream.readToEnd(teed[1]).then(result => {
       console.log(str + ': ', fn(result));
     });
-    return teed[0];
   },
 
   print_entire_stream_str: function (str, stream, fn = result => result) {
-    return util.print_entire_stream(str, stream, result => fn(util.Uint8Array_to_str(result)));
+    util.print_entire_stream(str, stream, result => fn(util.Uint8Array_to_str(result)));
   },
 
   getLeftNBits: function (array, bitcount) {

test/general/openpgp.js

@@ -1771,11 +1771,11 @@ describe('OpenPGP.js public api tests', function() {
             return openpgp.encrypt(encryptOpt).then(function (encrypted) {
                 decryptOpt.message = encrypted.message;
                 return encrypted.message.decrypt(decryptOpt.privateKeys);
-            }).then(function (packets) {
+            }).then(async function (packets) {
                 const literals = packets.packets.filterByTag(openpgp.enums.packet.literal);
                 expect(literals.length).to.equal(1);
                 expect(+literals[0].date).to.equal(+future);
-                expect(packets.getText()).to.equal(plaintext);
+                expect(await openpgp.stream.readToEnd(packets.getText())).to.equal(plaintext);
             });
         });
 
@@ -1796,11 +1796,11 @@ describe('OpenPGP.js public api tests', function() {
             return openpgp.encrypt(encryptOpt).then(function (encrypted) {
                 decryptOpt.message = encrypted.message;
                 return encrypted.message.decrypt(decryptOpt.privateKeys);
-            }).then(function (packets) {
+            }).then(async function (packets) {
                 const literals = packets.packets.filterByTag(openpgp.enums.packet.literal);
                 expect(literals.length).to.equal(1);
                 expect(+literals[0].date).to.equal(+past);
-                expect(packets.getLiteralData()).to.deep.equal(data);
+                expect(await openpgp.stream.readToEnd(packets.getLiteralData())).to.deep.equal(data);
             });
         });
 
@@ -1816,11 +1816,11 @@ describe('OpenPGP.js public api tests', function() {
 
             return openpgp.encrypt(encryptOpt).then(function (encrypted) {
                 return encrypted.message.decrypt(encryptOpt.privateKeys);
-            }).then(function (packets) {
+            }).then(async function (packets) {
                 const literals = packets.packets.filterByTag(openpgp.enums.packet.literal);
                 expect(literals.length).to.equal(1);
                 expect(+literals[0].date).to.equal(+past);
-                expect(packets.getText()).to.equal(plaintext);
+                expect(await openpgp.stream.readToEnd(packets.getText())).to.equal(plaintext);
                 return packets.verify(encryptOpt.publicKeys, past);
             }).then(function (signatures) {
                 expect(+signatures[0].signature.packets[0].created).to.equal(+past);
@@ -1844,12 +1844,12 @@ describe('OpenPGP.js public api tests', function() {
 
             return openpgp.encrypt(encryptOpt).then(function (encrypted) {
                 return encrypted.message.decrypt(encryptOpt.privateKeys);
-            }).then(function (packets) {
+            }).then(async function (packets) {
                 const literals = packets.packets.filterByTag(openpgp.enums.packet.literal);
                 expect(literals.length).to.equal(1);
                 expect(literals[0].format).to.equal('binary');
                 expect(+literals[0].date).to.equal(+future);
-                expect(packets.getLiteralData()).to.deep.equal(data);
+                expect(await openpgp.stream.readToEnd(packets.getLiteralData())).to.deep.equal(data);
                 return packets.verify(encryptOpt.publicKeys, future);
             }).then(function (signatures) {
                 expect(+signatures[0].signature.packets[0].created).to.equal(+future);
@@ -1874,12 +1874,12 @@ describe('OpenPGP.js public api tests', function() {
 
             return openpgp.encrypt(encryptOpt).then(function (encrypted) {
                 return encrypted.message.decrypt(encryptOpt.privateKeys);
-            }).then(function (packets) {
+            }).then(async function (packets) {
                 const literals = packets.packets.filterByTag(openpgp.enums.packet.literal);
                 expect(literals.length).to.equal(1);
                 expect(literals[0].format).to.equal('mime');
                 expect(+literals[0].date).to.equal(+future);
-                expect(packets.getLiteralData()).to.deep.equal(data);
+                expect(await openpgp.stream.readToEnd(packets.getLiteralData())).to.deep.equal(data);
                 return packets.verify(encryptOpt.publicKeys, future);
             }).then(function (signatures) {
                 expect(+signatures[0].signature.packets[0].created).to.equal(+future);

test/general/packet.js

@@ -218,8 +218,8 @@ describe("Packet", function() {
     randomBytesStub.returns(resolves(iv));
 
     return enc.encrypt(algo, key).then(async function() {
-      const [data, dataClone] = openpgp.stream.tee(msg.write());
+      const data = msg.write();
-      expect(await openpgp.stream.readToEnd(dataClone)).to.deep.equal(packetBytes);
+      expect(await openpgp.stream.readToEnd(openpgp.stream.clone(data))).to.deep.equal(packetBytes);
       await msg2.read(data);
       return msg2[0].decrypt(algo, key);
     }).then(async function() {
@@ -531,8 +531,8 @@ describe("Packet", function() {
       enc.packets.push(literal);
       await enc.encrypt(algo, key);
 
-      const [data, dataClone] = openpgp.stream.tee(msg.write());
+      const data = msg.write();
-      expect(await openpgp.stream.readToEnd(dataClone)).to.deep.equal(packetBytes);
+      expect(await openpgp.stream.readToEnd(openpgp.stream.clone(data))).to.deep.equal(packetBytes);
 
       const msg2 = new openpgp.packet.List();
       await msg2.read(data);
@@ -610,8 +610,8 @@ describe("Packet", function() {
       enc.packets.push(literal);
       await enc.encrypt(algo, key);
 
-      const [data, dataClone] = openpgp.stream.tee(msg.write());
+      const data = msg.write();
-      expect(await openpgp.stream.readToEnd(dataClone)).to.deep.equal(packetBytes);
+      expect(await openpgp.stream.readToEnd(openpgp.stream.clone(data))).to.deep.equal(packetBytes);
 
       const msg2 = new openpgp.packet.List();
       await msg2.read(data);

test/general/streaming.js

@@ -6,7 +6,7 @@ chai.use(require('chai-as-promised'));
 
 const { expect } = chai;
 
-const { Stream, util } = openpgp;
+const { util } = openpgp;
 
 describe('Streaming', function() {
   it('Encrypt small message', async function() {