<commit_before>var handshake = require('pull-handshake')
var crypto = require('crypto')
var through = require('pull-through')
var pull = require('pull-stream')
var deferredThrough = require('./resume')
var header = require('./header')
var peek = require('./peek')
var cipher = require('./cipher')

function isString(s) {
  return 'string' === typeof s
}

var m = 0
module.exports = function (dh, alg) {

  dh = dh || 'modp5'
  alg = alg || 'aes-256-cbc'
  return function (stream) {
    var k = ++m
    console.log('wrapStream', k)
    var encrypt = crypto.getDiffieHellman(dh)
    var decrypt = crypto.getDiffieHellman(dh)

    encrypt.generateKeys()
    decrypt.generateKeys()

    var len = encrypt.getPublicKey().length
    var deferred = pull.defer()

    var decryptS, encryptS

    var sEncrypt = cipher.encrypt(alg)
    var sDecrypt = cipher.decrypt(alg)

    var keys = Buffer.concat([
          encrypt.getPublicKey(),
          decrypt.getPublicKey()
        ])

    var i = 0, j = 0

    var seen = {}

    return {
      sink: pull(
        header(len*2, function (head) {
          sDecrypt.secret(decrypt.computeSecret(head.slice(0, len)))
          sEncrypt.secret(encrypt.computeSecret(head.slice(len, len*2)))
        }),
        peek(),
        pull.through(function (data) {
          var d = data.toString('hex')
          console.log('W', k, ++j, data.length, data.toString('hex'))
          if(seen[d]) throw new Error('already seen data:' + d)
          seen[d] = true
        }),
        sDecrypt,
        stream.sink
      ),
      source:
        pull(
          stream.source,
          sEncrypt,
          pull.prepend(keys),
          pull.through(function (data) {
            console.log('R', k, ++i, data.length, data.toString('hex'))
          })
        )
    }
  }
}
<commit_msg>rewrite<commit_after>var crypto = require('crypto')
var pull = require('pull-stream')
var deferredThrough = require('./resume')
var header = require('./header')
var peek = require('./peek')
var cipher = require('./cipher')

module.exports = function (dh, alg) {

  dh = dh || 'modp5'
  alg = alg || 'aes-256-cbc'
  return function (stream) {
    var encryptDH = crypto.getDiffieHellman(dh)
    var decryptDH = crypto.getDiffieHellman(dh)

    encryptDH.generateKeys()
    decryptDH.generateKeys()

    var length = encryptDH.getPublicKey().length

    var encrypt = cipher.encrypt(alg)
    var decrypt = cipher.decrypt(alg)

    var keys = Buffer.concat([
          encryptDH.getPublicKey(),
          decryptDH.getPublicKey()
        ])

    return {
      sink: pull(
        header(length*2, function (head) {
          decrypt.secret(decryptDH.computeSecret(head.slice(0, length)))
          encrypt.secret(encryptDH.computeSecret(head.slice(length, length*2)))
        }),
        peek(),
        decrypt,
        stream.sink
      ),
      source:
        pull(
          stream.source,
          encrypt,
          pull.prepend(keys)
        )
    }
  }
}

{'fraction_non_alphanumeric': 0.14137004701141706, 'fraction_numerical': 0.005708529214237743, 'mean_word_length': 2.3623024830699775, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2662609', 'n_tokens_mistral': 988, 'n_tokens_neox': 928, 'n_words': 201}