Support Node streams

src/message.js

@@ -96,6 +96,7 @@ Message.prototype.getSigningKeyIds = function() {
  * @param  {Array<Key>} privateKeys     (optional) private keys with decrypted secret data
  * @param  {Array<String>} passwords    (optional) passwords used to decrypt
  * @param  {Array<Object>} sessionKeys  (optional) session keys in the form: { data:Uint8Array, algorithm:String, [aeadAlgorithm:String] }
+ * @param  {Boolean} streaming          (optional) whether to process data as a stream
  * @returns {Promise<Message>}             new message with decrypted content
  * @async
  */
@@ -257,6 +258,7 @@ Message.prototype.getText = function() {
  * @param  {Boolean} wildcard          (optional) use a key ID of 0 instead of the public key IDs
  * @param  {Date} date                 (optional) override the creation date of the literal package
  * @param  {Object} userId             (optional) user ID to encrypt for, e.g. { name:'Robert Receiver', email:'robert@openpgp.org' }
+ * @param  {Boolean} streaming         (optional) whether to process data as a stream
  * @returns {Promise<Message>}                   new message with encrypted content
  * @async
  */
@@ -533,6 +535,7 @@ export async function createSignaturePackets(literalDataPacket, privateKeys, sig
  * Verify message signatures
  * @param {Array<module:key.Key>} keys array of keys to verify signatures
  * @param {Date} date (optional) Verify the signature against the given date, i.e. check signature creation time < date < expiration time
+ * @param  {Boolean} streaming (optional) whether to process data as a stream
  * @returns {Promise<Array<({keyid: module:type/keyid, valid: Boolean})>>} list of signer's keyid and validity of signature
  * @async
  */
@@ -683,8 +686,12 @@ Message.prototype.armor = function() {
 export async function readArmored(armoredText) {
   //TODO how do we want to handle bad text? Exception throwing
   //TODO don't accept non-message armored texts
+  const streamType = util.isStream(armoredText);
+  if (streamType === 'node') {
+    armoredText = stream.nodeToWeb(armoredText);
+  }
   const input = await armor.decode(armoredText);
-  return read(input.data, util.isStream(armoredText));
+  return read(input.data, streamType);
 }
 
 /**
@@ -695,7 +702,11 @@ export async function readArmored(armoredText) {
  * @async
  * @static
  */
-export async function read(input, fromStream) {
+export async function read(input, fromStream=util.isStream(input)) {
+  const streamType = util.isStream(input);
+  if (streamType === 'node') {
+    input = stream.nodeToWeb(input);
+  }
   const packetlist = new packet.List();
   await packetlist.read(input);
   const message = new Message(packetlist);
@@ -713,6 +724,10 @@ export async function read(input, fromStream) {
  * @static
  */
 export function fromText(text, filename, date=new Date(), type='utf8') {
+  const streamType = util.isStream(text);
+  if (streamType === 'node') {
+    text = stream.nodeToWeb(text);
+  }
   const literalDataPacket = new packet.Literal(date);
   // text will be converted to UTF8
   literalDataPacket.setText(text, type);
@@ -722,7 +737,7 @@ export function fromText(text, filename, date=new Date(), type='utf8') {
   const literalDataPacketlist = new packet.List();
   literalDataPacketlist.push(literalDataPacket);
   const message = new Message(literalDataPacketlist);
-  message.fromStream = util.isStream(text);
+  message.fromStream = streamType;
   return message;
 }
 
@@ -736,8 +751,12 @@ export function fromText(text, filename, date=new Date(), type='utf8') {
  * @static
  */
 export function fromBinary(bytes, filename, date=new Date(), type='binary') {
-  if (!util.isUint8Array(bytes) && !util.isStream(bytes)) {
+  const streamType = util.isStream(bytes);
-    throw new Error('Data must be in the form of a Uint8Array');
+  if (!util.isUint8Array(bytes) && !streamType) {
+    throw new Error('Data must be in the form of a Uint8Array or Stream');
+  }
+  if (streamType === 'node') {
+    bytes = stream.nodeToWeb(bytes);
   }
 
   const literalDataPacket = new packet.Literal(date);
@@ -748,6 +767,6 @@ export function fromBinary(bytes, filename, date=new Date(), type='binary') {
   const literalDataPacketlist = new packet.List();
   literalDataPacketlist.push(literalDataPacket);
   const message = new Message(literalDataPacketlist);
-  message.fromStream = util.isStream(bytes);
+  message.fromStream = streamType;
   return message;
 }

src/openpgp.js

@@ -281,7 +281,7 @@ export function encryptKey({ privateKey, passphrase }) {
  * @param  {Object} sessionKey                    (optional) session key in the form: { data:Uint8Array, algorithm:String }
  * @param  {module:enums.compression} compression (optional) which compression algorithm to compress the message with, defaults to what is specified in config
  * @param  {Boolean} armor                        (optional) if the return values should be ascii armored or the message/signature objects
- * @param  {Boolean} streaming                    (optional) whether to return data as a ReadableStream. Defaults to true if data is a Stream.
+ * @param  {'web'|'node'|false} streaming         (optional) whether to return data as a ReadableStream. Defaults to true if data is a Stream.
  * @param  {Boolean} detached                     (optional) if the signature should be detached (if true, signature will be added to returned object)
  * @param  {Signature} signature                  (optional) a detached signature to add to the encrypted message
  * @param  {Boolean} returnSessionKey             (optional) if the unencrypted session key should be added to returned object
@@ -339,7 +339,7 @@ export function encrypt({ message, publicKeys, privateKeys, passwords, sessionKe
  * @param  {Object|Array<Object>} sessionKeys (optional) session keys in the form: { data:Uint8Array, algorithm:String }
  * @param  {Key|Array<Key>} publicKeys        (optional) array of public keys or single key, to verify signatures
  * @param  {String} format                    (optional) return data format either as 'utf8' or 'binary'
- * @param  {Boolean} streaming                (optional) whether to return data as a ReadableStream. Defaults to true if message was created from a Stream.
+ * @param  {'web'|'node'|false} streaming     (optional) whether to return data as a ReadableStream. Defaults to true if message was created from a Stream.
  * @param  {Signature} signature              (optional) detached signature for verification
  * @param  {Date} date                        (optional) use the given date for verification instead of the current time
  * @returns {Promise<Object>}             decrypted and verified message in the form:
@@ -362,13 +362,10 @@ export function decrypt({ message, privateKeys, passwords, sessionKeys, publicKe
     const result = {};
     result.signatures = signature ? await decrypted.verifyDetached(signature, publicKeys, date, streaming) : await decrypted.verify(publicKeys, date, streaming);
     result.data = format === 'binary' ? decrypted.getLiteralData() : decrypted.getText();
-    result.data = await convertStream(result.data, streaming);
     result.filename = decrypted.getFilename();
-    if (streaming) {
+    if (streaming) linkStreams(result, message, decrypted.packets.stream);
-      linkStreams(result, message, decrypted.packets.stream);
+    result.data = await convertStream(result.data, streaming);
-    } else {
+    if (!streaming) await prepareSignatures(result.signatures);
-      await prepareSignatures(result.signatures);
-    }
     return result;
   }).catch(onError.bind(null, 'Error decrypting message'));
 }
@@ -386,7 +383,7 @@ export function decrypt({ message, privateKeys, passwords, sessionKeys, publicKe
  * @param  {CleartextMessage | Message} message (cleartext) message to be signed
  * @param  {Key|Array<Key>} privateKeys         array of keys or single key with decrypted secret key data to sign cleartext
  * @param  {Boolean} armor                      (optional) if the return value should be ascii armored or the message object
- * @param  {Boolean} streaming                  (optional) whether to return data as a ReadableStream. Defaults to true if data is a Stream.
+ * @param  {'web'|'node'|false} streaming       (optional) whether to return data as a ReadableStream. Defaults to true if data is a Stream.
  * @param  {Boolean} detached                   (optional) if the return value should contain a detached signature
  * @param  {Date} date                          (optional) override the creation date of the signature
  * @param  {Object} fromUserId                  (optional) user ID to sign with, e.g. { name:'Steve Sender', email:'steve@openpgp.org' }
@@ -425,13 +422,13 @@ export function sign({ message, privateKeys, armor=true, streaming=message&&mess
 
 /**
  * Verifies signatures of cleartext signed message
- * @param  {Key|Array<Key>} publicKeys   array of publicKeys or single key, to verify signatures
+ * @param  {Key|Array<Key>} publicKeys    array of publicKeys or single key, to verify signatures
- * @param  {CleartextMessage} message    cleartext message object with signatures
+ * @param  {CleartextMessage} message     cleartext message object with signatures
- * @param  {Boolean} streaming           (optional) whether to return data as a ReadableStream. Defaults to true if message was created from a Stream.
+ * @param  {'web'|'node'|false} streaming (optional) whether to return data as a ReadableStream. Defaults to true if message was created from a Stream.
- * @param  {Signature} signature         (optional) detached signature for verification
+ * @param  {Signature} signature          (optional) detached signature for verification
- * @param  {Date} date                   (optional) use the given date for verification instead of the current time
+ * @param  {Date} date                    (optional) use the given date for verification instead of the current time
  * @returns {Promise<Object>}             cleartext with status of verified signatures in the form of:
- *                                       { data:String, signatures: [{ keyid:String, valid:Boolean }] }
+ *                                        { data:String, signatures: [{ keyid:String, valid:Boolean }] }
  * @async
  * @static
  */
@@ -447,12 +444,9 @@ export function verify({ message, publicKeys, streaming=message&&message.fromStr
     const result = {};
     result.signatures = signature ? await message.verifyDetached(signature, publicKeys, date, streaming) : await message.verify(publicKeys, date, streaming);
     result.data = message instanceof CleartextMessage ? message.getText() : message.getLiteralData();
+    if (streaming) linkStreams(result, message);
     result.data = await convertStream(result.data, streaming);
-    if (streaming) {
+    if (!streaming) await prepareSignatures(result.signatures);
-      linkStreams(result, message);
-    } else {
-      await prepareSignatures(result.signatures);
-    }
     return result;
   }).catch(onError.bind(null, 'Error verifying cleartext signed message'));
 }
@@ -566,31 +560,34 @@ function toArray(param) {
 
 /**
  * Convert data to or from Stream
- * @param  {Object} data        the data to convert
+ * @param  {Object} data                   the data to convert
- * @param  {Boolean} streaming   (optional) whether to return a ReadableStream
+ * @param  {'web'|'node'|false} streaming  (optional) whether to return a ReadableStream
- * @returns {Object}            the data in the respective format
+ * @returns {Object}                       the data in the respective format
  */
 async function convertStream(data, streaming) {
   if (!streaming && util.isStream(data)) {
     return stream.readToEnd(data);
   }
   if (streaming && !util.isStream(data)) {
-    return new ReadableStream({
+    data = new ReadableStream({
       start(controller) {
         controller.enqueue(data);
         controller.close();
       }
     });
   }
+  if (streaming === 'node') {
+    data = stream.webToNode(data);
+  }
   return data;
 }
 
 /**
  * Convert object properties from Stream
- * @param  {Object} obj         the data to convert
+ * @param  {Object} obj                    the data to convert
- * @param  {Boolean} streaming   (optional) whether to return ReadableStreams
+ * @param  {'web'|'node'|false} streaming  (optional) whether to return ReadableStreams
- * @param  {Boolean} keys       (optional) which keys to return as streams, if possible
+ * @param  {Array<String>} keys            (optional) which keys to return as streams, if possible
- * @returns {Object}            the data in the respective format
+ * @returns {Object}                       the data in the respective format
  */
 async function convertStreams(obj, streaming, keys=[]) {
   if (Object.prototype.isPrototypeOf(obj)) {

src/packet/signature.js

@@ -600,10 +600,10 @@ Signature.prototype.toHash = function(data) {
   return util.concat([bytes, this.signatureData, this.calculateTrailer()]);
 };
 
-Signature.prototype.hash = function(data, toHash, asStream=true) {
+Signature.prototype.hash = function(data, toHash, streaming=true) {
   const hashAlgorithm = enums.write(enums.hash, this.hashAlgorithm);
   if (!toHash) toHash = this.toHash(data);
-  if (!asStream && util.isStream(toHash)) {
+  if (!streaming && util.isStream(toHash)) {
     return stream.fromAsync(async () => this.hash(data, await stream.readToEnd(toHash)));
   }
   return crypto.hash.digest(hashAlgorithm, toHash);

src/packet/sym_encrypted_aead_protected.js

@@ -91,15 +91,15 @@ SymEncryptedAEADProtected.prototype.write = function () {
  * Decrypt the encrypted payload.
  * @param  {String} sessionKeyAlgorithm   The session key's cipher algorithm e.g. 'aes128'
  * @param  {Uint8Array} key               The session key used to encrypt the payload
- * @param  {Boolean} asStream             Whether the top-level function will return a stream
+ * @param  {Boolean} streaming            Whether the top-level function will return a stream
  * @returns {Boolean}
  * @async
  */
-SymEncryptedAEADProtected.prototype.decrypt = async function (sessionKeyAlgorithm, key, asStream) {
+SymEncryptedAEADProtected.prototype.decrypt = async function (sessionKeyAlgorithm, key, streaming) {
   if (config.aead_protect_version !== 4) {
     this.cipherAlgo = enums.write(enums.symmetric, sessionKeyAlgorithm);
   }
-  await this.packets.read(await this.crypt('decrypt', key, stream.clone(this.encrypted), asStream));
+  await this.packets.read(await this.crypt('decrypt', key, stream.clone(this.encrypted), streaming));
   return true;
 };
 
@@ -107,17 +107,17 @@ SymEncryptedAEADProtected.prototype.decrypt = async function (sessionKeyAlgorith
  * Encrypt the packet list payload.
  * @param  {String} sessionKeyAlgorithm   The session key's cipher algorithm e.g. 'aes128'
  * @param  {Uint8Array} key               The session key used to encrypt the payload
- * @param  {Boolean} asStream             Whether the top-level function will return a stream
+ * @param  {Boolean} streaming            Whether the top-level function will return a stream
  * @async
  */
-SymEncryptedAEADProtected.prototype.encrypt = async function (sessionKeyAlgorithm, key, asStream) {
+SymEncryptedAEADProtected.prototype.encrypt = async function (sessionKeyAlgorithm, key, streaming) {
   this.cipherAlgo = enums.write(enums.symmetric, sessionKeyAlgorithm);
   this.aeadAlgo = config.aead_protect_version === 4 ? enums.write(enums.aead, this.aeadAlgorithm) : enums.aead.experimental_gcm;
   const mode = crypto[enums.read(enums.aead, this.aeadAlgo)];
   this.iv = await crypto.random.getRandomBytes(mode.ivLength); // generate new random IV
   this.chunkSizeByte = config.aead_chunk_size_byte;
   const data = this.packets.write();
-  this.encrypted = await this.crypt('encrypt', key, data, asStream);
+  this.encrypted = await this.crypt('encrypt', key, data, streaming);
 };
 
 /**
@@ -125,11 +125,11 @@ SymEncryptedAEADProtected.prototype.encrypt = async function (sessionKeyAlgorith
  * @param  {encrypt|decrypt} fn      Whether to encrypt or decrypt
  * @param  {Uint8Array} key          The session key used to en/decrypt the payload
  * @param  {Uint8Array | ReadableStream<Uint8Array>} data         The data to en/decrypt
- * @param  {Boolean} asStream        Whether the top-level function will return a stream
+ * @param  {Boolean} streaming        Whether the top-level function will return a stream
  * @returns {Uint8Array | ReadableStream<Uint8Array>}
  * @async
  */
-SymEncryptedAEADProtected.prototype.crypt = async function (fn, key, data, asStream) {
+SymEncryptedAEADProtected.prototype.crypt = async function (fn, key, data, streaming) {
   const cipher = enums.read(enums.symmetric, this.cipherAlgo);
   const mode = crypto[enums.read(enums.aead, this.aeadAlgo)];
   const modeInstance = await mode(cipher, key);
@@ -150,7 +150,7 @@ SymEncryptedAEADProtected.prototype.crypt = async function (fn, key, data, asStr
     return stream.transformPair(data, async (readable, writable) => {
       const reader = stream.getReader(readable);
       const buffer = new TransformStream({}, {
-        highWaterMark: asStream ? util.getHardwareConcurrency() * 2 ** (config.aead_chunk_size_byte + 6) : Infinity,
+        highWaterMark: streaming ? util.getHardwareConcurrency() * 2 ** (config.aead_chunk_size_byte + 6) : Infinity,
         size: array => array.length
       });
       stream.pipe(buffer.readable, writable);

src/packet/sym_encrypted_integrity_protected.js

@@ -87,13 +87,13 @@ SymEncryptedIntegrityProtected.prototype.write = function () {
  * Encrypt the payload in the packet.
  * @param  {String} sessionKeyAlgorithm   The selected symmetric encryption algorithm to be used e.g. 'aes128'
  * @param  {Uint8Array} key               The key of cipher blocksize length to be used
- * @param  {Boolean} asStream             Whether to set this.encrypted to a stream
+ * @param  {Boolean} streaming            Whether to set this.encrypted to a stream
  * @returns {Promise<Boolean>}
  * @async
  */
-SymEncryptedIntegrityProtected.prototype.encrypt = async function (sessionKeyAlgorithm, key, asStream) {
+SymEncryptedIntegrityProtected.prototype.encrypt = async function (sessionKeyAlgorithm, key, streaming) {
   let bytes = this.packets.write();
-  if (!asStream) bytes = await stream.readToEnd(bytes);
+  if (!streaming) bytes = await stream.readToEnd(bytes);
   const prefixrandom = await crypto.getPrefixRandom(sessionKeyAlgorithm);
   const repeat = new Uint8Array([prefixrandom[prefixrandom.length - 2], prefixrandom[prefixrandom.length - 1]]);
   const prefix = util.concat([prefixrandom, repeat]);
@@ -117,17 +117,17 @@ SymEncryptedIntegrityProtected.prototype.encrypt = async function (sessionKeyAlg
  * Decrypts the encrypted data contained in the packet.
  * @param  {String} sessionKeyAlgorithm   The selected symmetric encryption algorithm to be used e.g. 'aes128'
  * @param  {Uint8Array} key               The key of cipher blocksize length to be used
- * @param  {Boolean} asStream             Whether to read this.encrypted as a stream
+ * @param  {Boolean} streaming            Whether to read this.encrypted as a stream
  * @returns {Promise<Boolean>}
  * @async
  */
-SymEncryptedIntegrityProtected.prototype.decrypt = async function (sessionKeyAlgorithm, key, asStream) {
+SymEncryptedIntegrityProtected.prototype.decrypt = async function (sessionKeyAlgorithm, key, streaming) {
-  if (!asStream) this.encrypted = await stream.readToEnd(this.encrypted);
+  if (!streaming) this.encrypted = await stream.readToEnd(this.encrypted);
   const encrypted = stream.clone(this.encrypted);
   const encryptedClone = stream.passiveClone(encrypted);
   let decrypted;
   if (sessionKeyAlgorithm.substr(0, 3) === 'aes') { // AES optimizations. Native code for node, asmCrypto for browser.
-    decrypted = aesDecrypt(sessionKeyAlgorithm, encrypted, key, asStream);
+    decrypted = aesDecrypt(sessionKeyAlgorithm, encrypted, key, streaming);
   } else {
     decrypted = crypto.cfb.decrypt(sessionKeyAlgorithm, key, await stream.readToEnd(encrypted), false);
   }

test/general/openpgp.js

@@ -2216,7 +2216,7 @@ describe('OpenPGP.js public api tests', function() {
               message,
               format: 'binary'
             });
-            expect(openpgp.util.isStream(decrypted.data)).to.be.true;
+            expect(openpgp.util.isStream(decrypted.data)).to.equal('web');
             expect(await openpgp.stream.readToEnd(decrypted.data)).to.deep.equal(openpgp.util.concatUint8Array(plaintext));
           });
         });

test/general/streaming.js

@@ -76,7 +76,9 @@ const priv_key =
 
 const passphrase = 'hello world';
 
-describe('Streaming', function() {
+let plaintext, data, i, canceled, expectedType;
+
+function tests() {
   it('Encrypt small message', async function() {
     const data = new ReadableStream({
       async start(controller) {
@@ -99,26 +101,14 @@ describe('Streaming', function() {
   });
 
   it('Encrypt larger message', async function() {
-    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({
       message: openpgp.message.fromBinary(data),
       passwords: ['test'],
     });
-    expect(await openpgp.stream.getReader(openpgp.stream.clone(encrypted.data)).readBytes(1024)).to.match(/^-----BEGIN PGP MESSAGE-----\r\n/);
+    const reader = openpgp.stream.getReader(encrypted.data);
+    expect(await reader.peekBytes(1024)).to.match(/^-----BEGIN PGP MESSAGE-----\r\n/);
     if (i > 10) throw new Error('Data did not arrive early.');
+    reader.releaseLock();
     const msgAsciiArmored = await openpgp.stream.readToEnd(encrypted.data);
     const message = await openpgp.message.readArmored(msgAsciiArmored);
     const decrypted = await openpgp.decrypt({
@@ -130,24 +120,6 @@ describe('Streaming', function() {
   });
 
   it('Input stream should be canceled when canceling encrypted stream', async function() {
-    let plaintext = [];
-    let i = 0;
-    let canceled = false;
-    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();
-        }
-      },
-      cancel() {
-        canceled = true;
-      }
-    });
     const encrypted = await openpgp.encrypt({
       message: openpgp.message.fromBinary(data),
       passwords: ['test'],
@@ -164,24 +136,6 @@ describe('Streaming', function() {
     const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
     await privKey.decrypt(passphrase);
 
-    let plaintext = [];
-    let i = 0;
-    let canceled = false;
-    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();
-        }
-      },
-      cancel() {
-        canceled = true;
-      }
-    });
     const signed = await openpgp.sign({
       message: openpgp.message.fromBinary(data),
       privateKeys: privKey
@@ -195,19 +149,6 @@ describe('Streaming', function() {
   });
 
   it('Encrypt and decrypt larger message roundtrip', async function() {
-    let plaintext = [];
-    let i = 0;
-    const data = new ReadableStream({
-      async pull(controller) {
-        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({
       message: openpgp.message.fromBinary(data),
       passwords: ['test'],
@@ -220,30 +161,17 @@ describe('Streaming', function() {
       message,
       format: 'binary'
     });
-    expect(util.isStream(decrypted.data)).to.be.true;
+    expect(util.isStream(decrypted.data)).to.equal(expectedType);
-    expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).to.deep.equal(plaintext[0]);
+    const reader = openpgp.stream.getReader(decrypted.data);
+    expect(await reader.peekBytes(1024)).to.deep.equal(plaintext[0]);
     if (i <= 10) throw new Error('Data arrived early.');
-    expect(await openpgp.stream.readToEnd(decrypted.data)).to.deep.equal(util.concatUint8Array(plaintext));
+    expect(await reader.readToEnd()).to.deep.equal(util.concatUint8Array(plaintext));
   });
 
   it('Encrypt and decrypt larger message roundtrip (allow_unauthenticated_stream=true)', async function() {
     let allow_unauthenticated_streamValue = openpgp.config.allow_unauthenticated_stream;
     openpgp.config.allow_unauthenticated_stream = true;
     try {
-      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({
         message: openpgp.message.fromBinary(data),
         passwords: ['test'],
@@ -256,11 +184,12 @@ describe('Streaming', function() {
         message,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
       expect(util.isStream(decrypted.signatures)).to.be.false;
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(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));
+      expect(await reader.readToEnd()).to.deep.equal(util.concatUint8Array(plaintext));
       expect(decrypted.signatures).to.exist.and.have.length(0);
     } finally {
       openpgp.config.allow_unauthenticated_stream = allow_unauthenticated_streamValue;
@@ -275,20 +204,6 @@ describe('Streaming', function() {
       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({
         message: openpgp.message.fromBinary(data),
         publicKeys: pubKey,
@@ -303,10 +218,11 @@ describe('Streaming', function() {
         message,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(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));
+      expect(await reader.readToEnd()).to.deep.equal(util.concatUint8Array(plaintext));
     } finally {
       openpgp.config.allow_unauthenticated_stream = allow_unauthenticated_streamValue;
     }
@@ -316,20 +232,6 @@ describe('Streaming', function() {
     let allow_unauthenticated_streamValue = openpgp.config.allow_unauthenticated_stream;
     openpgp.config.allow_unauthenticated_stream = true;
     try {
-      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({
         message: openpgp.message.fromBinary(data),
         passwords: ['test'],
@@ -344,12 +246,14 @@ describe('Streaming', function() {
       const decrypted = await openpgp.decrypt({
         passwords: ['test'],
         message,
+        streaming: expectedType,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).not.to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(1024)).not.to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
-      await expect(openpgp.stream.readToEnd(decrypted.data)).to.be.rejectedWith('Modification detected.');
+      await expect(reader.readToEnd()).to.be.rejectedWith('Modification detected.');
       expect(decrypted.signatures).to.exist.and.have.length(0);
     } finally {
       openpgp.config.allow_unauthenticated_stream = allow_unauthenticated_streamValue;
@@ -364,20 +268,6 @@ describe('Streaming', function() {
       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(resolve => setTimeout(resolve, 100));
-          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({
         message: openpgp.message.fromBinary(data),
         publicKeys: pubKey,
@@ -393,12 +283,14 @@ describe('Streaming', function() {
         publicKeys: pubKey,
         privateKeys: privKey,
         message,
+        streaming: expectedType,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(10)).not.to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(10)).not.to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
-      await expect(openpgp.stream.readToEnd(decrypted.data)).to.be.rejectedWith('Ascii armor integrity check on message failed');
+      await expect(reader.readToEnd()).to.be.rejectedWith('Ascii armor integrity check on message failed');
       expect(decrypted.signatures).to.exist.and.have.length(1);
     } finally {
       openpgp.config.allow_unauthenticated_stream = allow_unauthenticated_streamValue;
@@ -413,20 +305,6 @@ describe('Streaming', function() {
       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(resolve => setTimeout(resolve, 100));
-          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({
         message: openpgp.message.fromBinary(data),
         publicKeys: pubKey,
@@ -441,12 +319,14 @@ describe('Streaming', function() {
       const decrypted = await openpgp.decrypt({
         privateKeys: privKey,
         message,
+        streaming: expectedType,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(10)).not.to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(10)).not.to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
-      await expect(openpgp.stream.readToEnd(decrypted.data)).to.be.rejectedWith('Ascii armor integrity check on message failed');
+      await expect(reader.readToEnd()).to.be.rejectedWith('Ascii armor integrity check on message failed');
       expect(decrypted.signatures).to.exist.and.have.length(1);
       expect(await decrypted.signatures[0].verified).to.be.null;
     } finally {
@@ -462,20 +342,6 @@ describe('Streaming', function() {
       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(resolve => setTimeout(resolve, 100));
-          if (i++ < 10) {
-            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-            controller.enqueue(randomBytes);
-            plaintext.push(randomBytes);
-          } else {
-            controller.close();
-          }
-        }
-      });
       const signed = await openpgp.sign({
         message: openpgp.message.fromBinary(data),
         privateKeys: privKey
@@ -488,12 +354,14 @@ describe('Streaming', function() {
       }));
       const verified = await openpgp.verify({
         publicKeys: pubKey,
-        message
+        message,
+        streaming: expectedType
       });
-      expect(util.isStream(verified.data)).to.be.true;
+      expect(util.isStream(verified.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(verified.data)).readBytes(10)).not.to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(verified.data);
+      expect(await reader.peekBytes(10)).not.to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
-      await expect(openpgp.stream.readToEnd(verified.data)).to.be.rejectedWith('Ascii armor integrity check on message failed');
+      await expect(reader.readToEnd()).to.be.rejectedWith('Ascii armor integrity check on message failed');
       expect(verified.signatures).to.exist.and.have.length(1);
     } finally {
       openpgp.config.allow_unauthenticated_stream = allow_unauthenticated_streamValue;
@@ -506,20 +374,6 @@ describe('Streaming', function() {
     openpgp.config.aead_protect = true;
     openpgp.config.aead_chunk_size_byte = 4;
     try {
-      let plaintext = [];
-      let i = 0;
-      const data = new ReadableStream({
-        async pull(controller) {
-          await new Promise(resolve => setTimeout(resolve, 10));
-          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({
         message: openpgp.message.fromBinary(data),
         passwords: ['test'],
@@ -532,10 +386,11 @@ describe('Streaming', function() {
         message,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(1024)).to.deep.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect(await reader.peekBytes(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));
+      expect(await reader.readToEnd()).to.deep.equal(util.concatUint8Array(plaintext));
     } finally {
       openpgp.config.aead_protect = aead_protectValue;
       openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
@@ -564,6 +419,7 @@ describe('Streaming', function() {
       });
       const encrypted = await openpgp.encrypt({
         message: openpgp.message.fromText(data),
+        streaming: expectedType,
         passwords: ['test'],
       });
 
@@ -573,10 +429,11 @@ describe('Streaming', function() {
         passwords: ['test'],
         message
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
-      expect(await openpgp.stream.getReader(openpgp.stream.clone(decrypted.data)).readBytes(50)).to.equal(plaintext[0]);
+      const reader = openpgp.stream.getReader(decrypted.data);
+      expect((await reader.peekBytes(200)).toString('utf8').substr(0, 50)).to.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
-      expect(await openpgp.stream.readToEnd(decrypted.data)).to.equal(util.concat(plaintext));
+      expect((await reader.readToEnd()).toString('utf8')).to.equal(util.concat(plaintext));
     } finally {
       openpgp.config.aead_protect = aead_protectValue;
       openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
@@ -584,25 +441,6 @@ describe('Streaming', function() {
   });
 
   it('stream.transformPair()', async function() {
-    let plaintext = [];
-    let i = 0;
-    let canceled = false;
-    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();
-        }
-      },
-      cancel() {
-        canceled = true;
-      }
-    });
-
     const transformed = stream.transformPair(stream.slice(data, 0, 5000), async (readable, writable) => {
       const reader = stream.getReader(readable);
       const writer = stream.getWriter(writable);
@@ -631,24 +469,6 @@ describe('Streaming', function() {
     openpgp.config.aead_protect = true;
     openpgp.config.aead_chunk_size_byte = 4;
     try {
-      let plaintext = [];
-      let i = 0;
-      let canceled = false;
-      const data = new ReadableStream({
-        async pull(controller) {
-          await new Promise(resolve => setTimeout(resolve, 10));
-          if (i++ < 10) {
-            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-            controller.enqueue(randomBytes);
-            plaintext.push(randomBytes);
-          } else {
-            controller.close();
-          }
-        },
-        cancel() {
-          canceled = true;
-        }
-      });
       const encrypted = await openpgp.encrypt({
         message: openpgp.message.fromBinary(data),
         passwords: ['test'],
@@ -661,7 +481,7 @@ describe('Streaming', function() {
         message,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
       const reader = openpgp.stream.getReader(decrypted.data);
       expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
@@ -684,24 +504,6 @@ describe('Streaming', function() {
       const privKey = (await openpgp.key.readArmored(priv_key)).keys[0];
       await privKey.decrypt(passphrase);
 
-      let plaintext = [];
-      let i = 0;
-      let canceled = false;
-      const data = new ReadableStream({
-        async pull(controller) {
-          await new Promise(resolve => setTimeout(resolve, 10));
-          if (i++ < 10) {
-            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-            controller.enqueue(randomBytes);
-            plaintext.push(randomBytes);
-          } else {
-            controller.close();
-          }
-        },
-        cancel() {
-          canceled = true;
-        }
-      });
       const signed = await openpgp.sign({
         message: openpgp.message.fromBinary(data),
         privateKeys: privKey
@@ -713,7 +515,7 @@ describe('Streaming', function() {
         publicKeys: pubKey,
         message
       });
-      expect(util.isStream(verified.data)).to.be.true;
+      expect(util.isStream(verified.data)).to.equal(expectedType);
       const reader = openpgp.stream.getReader(verified.data);
       expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
@@ -729,20 +531,6 @@ describe('Streaming', function() {
   });
 
   it("Don't pull entire input stream when we're not pulling encrypted stream", async function() {
-    let plaintext = [];
-    let i = 0;
-    const data = new ReadableStream({
-      async pull(controller) {
-        if (i++ < 100) {
-          let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-          controller.enqueue(randomBytes);
-          plaintext.push(randomBytes);
-        } else {
-          controller.close();
-        }
-        await new Promise(setTimeout);
-      }
-    });
     const encrypted = await openpgp.encrypt({
       message: openpgp.message.fromBinary(data),
       passwords: ['test'],
@@ -759,20 +547,6 @@ describe('Streaming', function() {
     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) {
-        if (i++ < 100) {
-          let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-          controller.enqueue(randomBytes);
-          plaintext.push(randomBytes);
-        } else {
-          controller.close();
-        }
-        await new Promise(setTimeout);
-      }
-    });
     const signed = await openpgp.sign({
       message: openpgp.message.fromBinary(data),
       privateKeys: privKey
@@ -792,20 +566,6 @@ describe('Streaming', function() {
     let coresStub = stub(openpgp.util, 'getHardwareConcurrency');
     coresStub.returns(1);
     try {
-      let plaintext = [];
-      let i = 0;
-      const data = new ReadableStream({
-        async pull(controller) {
-          if (i++ < 100) {
-            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-            controller.enqueue(randomBytes);
-            plaintext.push(randomBytes);
-          } else {
-            controller.close();
-          }
-          await new Promise(setTimeout);
-        }
-      });
       const encrypted = await openpgp.encrypt({
         message: openpgp.message.fromBinary(data),
         passwords: ['test'],
@@ -817,7 +577,7 @@ describe('Streaming', function() {
         message,
         format: 'binary'
       });
-      expect(util.isStream(decrypted.data)).to.be.true;
+      expect(util.isStream(decrypted.data)).to.equal(expectedType);
       const reader = openpgp.stream.getReader(decrypted.data);
       expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
@@ -840,20 +600,6 @@ describe('Streaming', function() {
       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) {
-          if (i++ < 100) {
-            let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
-            controller.enqueue(randomBytes);
-            plaintext.push(randomBytes);
-          } else {
-            controller.close();
-          }
-          await new Promise(setTimeout);
-        }
-      });
       const signed = await openpgp.sign({
         message: openpgp.message.fromBinary(data),
         privateKeys: privKey
@@ -864,7 +610,7 @@ describe('Streaming', function() {
         publicKeys: pubKey,
         message
       });
-      expect(util.isStream(verified.data)).to.be.true;
+      expect(util.isStream(verified.data)).to.equal(expectedType);
       const reader = openpgp.stream.getReader(verified.data);
       expect(await reader.readBytes(1024)).to.deep.equal(plaintext[0]);
       if (i > 10) throw new Error('Data did not arrive early.');
@@ -875,4 +621,70 @@ describe('Streaming', function() {
       openpgp.config.aead_chunk_size_byte = aead_chunk_size_byteValue;
     }
   });
+
+  if (openpgp.util.detectNode()) {
+    const fs = util.nodeRequire('fs');
+
+    it('Node: Encrypt and decrypt binary message roundtrip', async function() {
+      let plaintext = fs.readFileSync(__filename);
+      const data = fs.createReadStream(__filename);
+      const encrypted = await openpgp.encrypt({
+        message: openpgp.message.fromBinary(data),
+        passwords: ['test'],
+      });
+
+      const msgAsciiArmored = encrypted.data;
+      const message = await openpgp.message.readArmored(msgAsciiArmored);
+      const decrypted = await openpgp.decrypt({
+        passwords: ['test'],
+        message,
+        format: 'binary'
+      });
+      expect(util.isStream(decrypted.data)).to.equal('node');
+      expect(await openpgp.stream.readToEnd(decrypted.data)).to.deep.equal(plaintext);
+    });
+
+  }
+}
+
+describe('Streaming', function() {
+  let currentTest = 0;
+
+  beforeEach(function() {
+    let test = ++currentTest;
+
+    plaintext = [];
+    i = 0;
+    canceled = false;
+    data = new ReadableStream({
+      async pull(controller) {
+        await new Promise(setTimeout);
+        if (test === currentTest && i++ < 10) {
+          let randomBytes = await openpgp.crypto.random.getRandomBytes(1024);
+          controller.enqueue(randomBytes);
+          plaintext.push(randomBytes);
+        } else {
+          controller.close();
+        }
+      },
+      cancel() {
+        canceled = true;
+      }
+    });
+  });
+
+  tryTests('WhatWG Streams', tests, {
+    if: true,
+    beforeEach: function() {
+      expectedType = 'web';
+    }
+  });
+
+  tryTests('Node Streams', tests, {
+    if: openpgp.util.detectNode(),
+    beforeEach: function() {
+      data = openpgp.stream.webToNode(data);
+      expectedType = 'node';
+    }
+  });
 });