'use strict' const request = require('request') const socks_client = require('socks').SocksClient const fs = require('fs') const torm = require('./tor.js') const logm = require('./log.js') const worker = require('./worker.js') const measure_interval = 1000 const measure_count = 600 const parralelism = 100 const base_url = process.env.SERVER_URL || "http://127.0.0.1:3000" const repeat_for = (times, delay, fn) => new Promise((resolve, reject) => { const timer = setInterval(() => { if (times-- <= 0) { clearInterval(timer) resolve() return } try { fn() } catch(err) { clearInterval(timer) reject(err) return } }, delay) }) const doMeasure = socket => () => { socket.write(`${Date.now().toString()}\n`) } const clean = (id) => new Promise((resolve, reject) => { request({uri: `${base_url}/hidden_service`, method: "DELETE", json: {id: id}}, (err,res,body) => { if (err) reject(err) if (body.err) reject(body.err) resolve(body) }) }) const wait_for_proxy = (socks_opt, max) => new Promise((resolve, reject) => { const loop = cur => { socks_client.createConnection(socks_opt, (err, info) => { if (err && max <= 0) { console.error("No more retry") reject("No more retry") return } if (err) { console.error(`Proxy failed to build with error "${err.message}", ${cur} retries left...`) setTimeout(loop, 5000, cur - 1) return } resolve(info) }) } loop(max) }) const create_hidden_service = base_url => new Promise((resolve, reject) => { request.post(`${base_url}/hidden_service`, {json: {}}, (err,res,body) => { if (err) { reject(err) return } if (!body.hs || !body.port) reject("Missing value from API endpoint") else resolve(body) }) }) const wait_for_hs = socks_port => create_hidden_service(base_url) .then(body => Promise.all([ body, logm.get_write_fd(`logs/${Date.now()}-${body.hs}.onion-client.log`) ])) .then(([body, log]) => { fs.write(log, `#info ${Date.now().toString()} Got ${body.hs}:${body.port}\n`,() => {}) console.info(`We got this hidden service: ${body.hs} on port ${body.port}`) const socks_opt = { proxy: {ipaddress: '127.0.0.1', port: socks_port, type: 5}, command: 'connect', destination: { host: `${body.hs}.onion`, port: body.port } } /* // used to speedup tests socks_opt.destination.host = "ysw5uau3pbjaviha.onion" socks_opt.destination.port = 43035 */ return Promise.all([ wait_for_proxy(socks_opt, 120), {body: body, socks_opt: socks_opt, log: log} ]) }) .then(([info, data]) => new Promise((resolve, reject) => { info.socket.on('close', () => resolve(data)) info.socket.destroy() })) const do_measurement = data => Promise.all([wait_for_proxy(data.socks_opt), data]) .then(([info, data]) => { const socks_opt = data.socks_opt fs.write(data.log, `#info ${Date.now().toString()} Socks has been created, HS is up\n`, () => {}) console.info(`SOCKS proxy to ${socks_opt.destination.host}:${socks_opt.destination.port} has been created`) const recv_back = new Promise((resolve, reject) => { setTimeout(reject, measure_count * measure_interval + 2 * 60 * 1000, "Receive answer timed out") let missing = measure_count info.socket.on('data', recv_data => { const tss = recv_data.split('\n') tss.pop() // Remove last empty line tss.forEach((ts, idx) => { fs.write(data.log, `delta_timestamp, ${missing}, ${socks_opt.destination.host}, ${socks_opt.destination.port}, ${Date.now()}, ${ts}\n`, () => {}) if (idx > 0) fs.write(data.log, `#info aggregation: ${missing} was aggregated with ${missing + idx} for ${socks_opt.destination.host}\n`, () => {}) if (--missing == 0) resolve() }) }) }) info.socket.setNoDelay(true) info.socket.setEncoding("utf-8") data.info = info return Promise.all([ recv_back, repeat_for(measure_count, measure_interval, doMeasure(info.socket)), data ]) }) .then(([recv_back, repeat, data]) => { data.info.socket.destroy() fs.write(data.log, `#info ${Date.now().toString()} Done\n`, () => fs.close(data.log, () => {})) console.info(`Measurement terminated for ${data.socks_opt.destination.host}\n`) return Promise.all([clean(data.body.id), data]) }) .then(([cl, data]) => { console.info(`Done for ${data.socks_opt.destination.host}`) }) .catch(err => { data.info.socket.destroy() fs.write(data.log, `#info ${Date.now().toString()} ERROR occured\n`, () => fs.close(data.log, () => {})) console.error(err) }) const wait_bootstrap = tor => new Promise((resolve, reject) => { const loop = () => { console.info("Waiting for bootstrap, next check in 5s...") setTimeout(() => { tor.getInfo("status/circuit-established", (err, res) => { if (err) reject(err) else if (res == '0\n') loop() else resolve() }) }, 5000) } loop() }) const get_socks_port = tor => new Promise((resolve, reject) => { tor.getInfo('net/listeners/socks', (err, result) => { if (err) { reject(err) return } const port = parseInt(result.match(/:(\d+)/)[1]) console.info(`Tor daemon accepts SOCKS connections on port ${port}`) resolve(port) }) }) logm .init .then(([stream_fd, circ_fd]) => torm.init(stream_fd, circ_fd)) .then(tor => Promise.all([ get_socks_port(tor), wait_bootstrap(tor), tor ])) //.then(([socks_port, boot, tor]) => one_req(socks_port)) .then(([socks_port, boot, tor]) => { const hs_spawn_worker = new worker(200, 5000) const measurement_worker = new worker(100, 1000) for (let i = 0; i < 200; i++) { hs_spawn_worker.add(() => wait_for_hs(socks_port)) } hs_spawn_worker.on('success', v => { console.debug("HS ready, forwarding it to measurement") measurement_worker.add(() => do_measurement(v)) }) hs_spawn_worker.on('error', e => { console.error("Failed to create HS", e) hs_spawn_worker.add(() => wait_for_hs(socks_port)) }) measurement_worker.on('success', v => { console.debug("Measurement done, readding token to HS") hs_spawn_worker.add(() => wait_for_hs(socks_port)) }) measurement_worker.on('error', e => { console.error("Failed to do a measurement, readding token to HS") hs_spawn_worker.add(() => wait_for_hs(socks_port)) }) //let loop = () => one_req(socks_port).then(loop).catch(console.error) //for (let i = 0; i < 100; i++) setTimeout(loop, i * 10 * 1000) }) .catch(console.error) // status/bootstrap-phase // status/circuit-established // status/enough-dir-info // status/good-server-descriptor // status/accepted-server-descriptor