onion get wip continued

src/index.js

@@ -20,28 +20,28 @@ function timestamp() {
   return [d.substr(5, 5), d.substr(11, 8)].join(' ')
 }
 
-const logger = {
+class Logger {
   debug(...args) {
     if (LogLevels.debug > logLevel) return;
     const [format, ...rest] = args;
     console.debug(timestamp(), "D", util.format(format, ...rest));
-  },
+  }
   info(...args) {
     if (LogLevels.info > logLevel) return;
     const [format, ...rest] = args;
     console.debug(timestamp(), "I", util.format(format, ...rest));
-  },
+  }
   warn(...args) {
     if (LogLevels.warn > logLevel) return;
     const [format, ...rest] = args;
     console.debug(timestamp(), "W", util.format(format, ...rest));
-  },
+  }
   error(...args) {
     if (LogLevels.error > logLevel) return;
     const [format, ...rest] = args;
     console.debug(timestamp(), "E", util.format(format, ...rest));
-  },
+  }
-};
+}
 
 (async function main(args) {
   let nb_hops = 3;
@@ -69,6 +69,8 @@ const logger = {
     uri.port = (uri.protocol === 'https:' ? 443 : 80).toString();
   }
 
+  const logger = new Logger();
+
   /** @type Socket */
   let socket;
   /** @type Circuit */

src/tor/Circuit.js

@@ -3,6 +3,8 @@ const CircuitNode = require('./CircuitNode');
 const Cell = require('./Cell')
 const RelayCell = require('./RelayCell')
 const Stream = require('./TorStream')
+const TorStream = require('./TorStream')
+const { hybrid_encrypt } = require('../utils/crypto')
 const {sha1} = require('../utils/crypto')
 const { parseIp } = require('../utils/net')
 const { defer } = require('../utils/time')
@@ -31,12 +33,18 @@ class Circuit {
   _node_list = [];
   /** @type Socket */
   _socket = null;
+  /** @type number */
   _circuit_id;
-  _waits = {};
   /** @type CircuitNode */
   _extend_node = null;
+  /** @type {{[string]:TorStream}} */
   _stream_map = {};
+  _waits = {};
 
+  /**
+   * @param {Socket} socket
+   * @param {Logger} logger
+   */
   constructor (socket, logger) {
     this._socket = socket;
     this._circuit_id = (Circuit._next_circuit_id++) | 0x80000000;
@@ -55,8 +63,6 @@ class Circuit {
     return this._socket;
   }
 
-  get state() { return this._state };
-
   set state(state) {
     this._state = state;
     const waiters = this._waits[state];
@@ -90,7 +96,7 @@ class Circuit {
   }
 
   destroy() {
-    if (this.state === States.destroyed)
+    if (this._state === States.destroyed)
     {
       return;
     }
@@ -276,6 +282,7 @@ class Circuit {
 
   _on_relay_extended_cell(cell) {
     //TODO
+    throw new Error('Not implemented')
   }
 
   _on_relay_extended2(cell) {
@@ -383,6 +390,11 @@ class Circuit {
     }
   }
 
+  /**
+   * @param {string} host
+   * @param {number} port
+   * @returns {Promise<TorStream>}
+   */
   async create_stream(host, port) {
     //
     // tor-spec.txt
@@ -420,22 +432,55 @@ class Circuit {
     else
     {
       this._logger.error("circuit::create_stream() [is_ready() == false]");
+      return null;
     }
   }
 
   async create_onion_stream(onion, port) {
-    const hidden_service_connector = new HiddenServiceConnector(this, onion);
+    const hidden_service_connector = new HiddenServiceConnector(this, onion, this._logger);
 
     return await hidden_service_connector.connect()
-      ? this.create_stream(onion, port)
+      ? await this.create_stream(onion, port)
       : null;
   }
 
-  create_dir_stream() {
+  async create_dir_stream() {
-    //TODO
+    const stream_id = Circuit._next_stream_id++;
-    throw new Error('Not implemented');
+
+    const stream = new TorStream(stream_id, this, this._logger);
+    this._stream_map[stream_id] = stream;
+
+    this._logger.debug("Circuit.create_dir_stream() [stream: %i, state: connecting]", stream_id);
+    this.state = States.connecting;
+
+    this.send_relay_cell(stream_id, Cell.commands.relay_begin_dir);
+
+
+    if (await this.wait_for_state(States.ready))
+    {
+      this._logger.debug("Circuit.create_dir_stream() [stream: %i, state: connected]", stream_id);
+    }
+    else
+    {
+      this._logger.error("Circuit.create_dir_stream() [is_ready() == false]");
+    }
+
+    //
+    // if the circuit has been destroyed,
+    // the stream has been destroyed as well,
+    // so we don't need to delete it here.
+    //
+
+    return this.is_ready()
+      ? stream
+      : null;
   }
 
+  /**
+   * @param {OnionRouter} or
+   * @param {string} handshake_type
+   * @returns {Promise<void>}
+   */
   extend(or, handshake_type = 'ntor') {
     switch (handshake_type) {
       case 'ntor':
@@ -445,6 +490,11 @@ class Circuit {
     }
   }
 
+  /**
+   * @param {OnionRouter} next_onion_router
+   * @returns {Promise<void>}
+   * @private
+   */
   async _extend_ntor(next_onion_router) {
     //
     // An EXTEND2 cell's relay payload contains:
@@ -478,7 +528,7 @@ class Circuit {
     const ipv6 = 1;
     const legacy_id = 2;
 
-    this._logger.debug(`circuit::extend_ntor() [or: ${next_onion_router.name}, state: extending]`);
+    this._logger.debug(`Circuit._extend_ntor() [or: ${next_onion_router.name}, state: extending]`);
     this.state = States.extending;
 
     this._extend_node = new CircuitNode(this, next_onion_router, 'normal');
@@ -521,11 +571,11 @@ class Circuit {
 
     if (await this.wait_for_state(States.ready))
     {
-      this._logger.debug(`circuit::extend_ntor() [or: ${next_onion_router.name}, state: extended]`);
+      this._logger.debug(`Circuit._extend_ntor() [or: ${next_onion_router.name}, state: extended]`);
     }
     else
     {
-      this._logger.error(`circuit::extend_ntor() [or: ${next_onion_router.name}, state: destroyed]`);
+      this._logger.error(`Circuit._extend_ntor() [or: ${next_onion_router.name}, state: destroyed]`);
     }
   }
 
@@ -536,14 +586,22 @@ class Circuit {
     return this._node_list[this._node_list.length - 1];
   }
 
+  /**
+   * @param {number} stream_id
+   * @returns {TorStream}
+   */
   get_stream_by_id(stream_id) {
-    return this._stream_map[stream_id];
+    return this._stream_map[stream_id] || null;
   }
 
   send_relay_sendme_cell() {
     //TODO
+    throw new Error('Not implemented');
   }
 
+  /**
+   * @param {TorStream} stream
+   */
   send_relay_end_cell(stream) {
     this.send_relay_cell(
       stream.stream_id,
@@ -558,6 +616,10 @@ class Circuit {
     delete this._stream_map[stream.stream_id];
   }
 
+  /**
+   * @param {TorStream} stream
+   * @param {Buffer} data
+   */
   send_relay_data_cell(stream, data) {
     for (let i = 0; i < data.length; i += Cell.payload_size)
     {
@@ -572,10 +634,23 @@ class Circuit {
     }
   }
 
-  send_relay_cell(stream_id, relay_command, payload, cell_command = Cell.commands.relay_early, node = null) {
+  /**
+   * @param {number} stream_id
+   * @param {number} relay_command
+   * @param {Buffer} payload
+   * @param {number} cell_command
+   * @param {CircuitNode} node
+   */
+  send_relay_cell(
+    stream_id,
+    relay_command,
+    payload = Buffer.alloc(0),
+    cell_command = Cell.commands.relay_early,
+    node = null
+  ) {
     node = node ? node : this.get_final_circuit_node();
 
-    if (this.get_stream_by_id(stream_id) == null && stream_id !== 0)
+    if (this.get_stream_by_id(stream_id) === null && stream_id !== 0)
     {
       this._logger.warn("Circuit.send_relay_cell() attempt to send cell to non-existent stream-id:", stream_id);
       return;
@@ -590,13 +665,15 @@ class Circuit {
       Cell.relay_commands_lookup[relay_command],
       payload);
 
-    this._socket.send_cell(this._encrypt(new RelayCell(
+    this._socket.send_cell(
-      this._circuit_id,
+      this._encrypt(
-      cell_command,
+        new RelayCell(
-      node,
+          this._circuit_id,
-      relay_command,
+          cell_command,
-      stream_id,
+          node,
-      payload)));
+          relay_command,
+          stream_id,
+          payload)));
   }
 
   /**
@@ -625,6 +702,11 @@ class Circuit {
     return new RelayCell();
   }
 
+  /**
+   * @param {number} desired_state
+   * @param {number} [timeout]
+   * @returns {Promise<boolean>}
+   */
   wait_for_state(desired_state, timeout = 30000) {
     const d = defer();
     d.timeout = setTimeout(() => {
@@ -635,21 +717,47 @@ class Circuit {
     return d.promise;
   }
 
+  /**
+   * @param {Buffer} rendezvous_cookie
+   * @returns {Promise<void>}
+   */
   async rendezvous_establish(rendezvous_cookie) {
-    //TODO
+    //mini_assert(rendezvous_cookie.get_size() == 20);
-    throw new Error("Not implemented");
+
+    this._logger.debug("Circuit.rendezvous_establish() [circuit: %i, state: establishing]", this._circuit_id);
+    this.state = States.rendezvous_establishing;
+
+    this.send_relay_cell(
+      0,
+      Cell.commands.relay_command_establish_rendezvous,
+      rendezvous_cookie);
+
+
+    if (await this.wait_for_state(States.rendezvous_established))
+    {
+      this._logger.debug("Circuit.rendezvous_establish() [circuit: %i, state: established]", this._circuit_id);
+    }
+    else
+    {
+      this._logger.error("Circuit.rendezvous_establish() [circuit: %i, is_rendezvous_established() == false]", this._circuit_id);
+    }
   }
 
+  /**
+   * @param {Circuit} rendezvous_circuit
+   * @param {Buffer} rendezvous_cookie
+   * @returns {Promise<void>}
+   */
   async rendezvous_introduce(rendezvous_circuit, rendezvous_cookie) {
     //mini_assert(rendezvous_cookie.get_size() == 20);
 
     const introduction_point = this.get_final_circuit_node().onion_router;
     const introducee = rendezvous_circuit.get_final_circuit_node.onion_router;
 
-    this._logger.debug("circuit::rendezvous_introduce() [or: %s, state: introducing]", introduction_point.name);
+    this._logger.debug("Circuit.rendezvous_introduce() [or: %s, state: introducing]", introduction_point.name);
     this.state = States.rendezvous_introducing;
 
-    this._logger.debug("circuit::rendezvous_introduce() [or: %s, state: completing]", introduction_point.name);
+    this._logger.debug("Circuit.rendezvous_introduce() [or: %s, state: completing]", introduction_point.name);
     rendezvous_circuit.state = States.rendezvous_completing;
 
     //
@@ -689,20 +797,18 @@ class Circuit {
     //io::memory_stream handshake_stream(handshake_bytes);
     //io::stream_wrapper handshake_buffer(handshake_stream, endianness::big_endian);
 
-    rendezvous_circuit._extend_node = new CircuitNode(this, introduction_point, circuit_node_type::introduction_point);
+    rendezvous_circuit._extend_node = new CircuitNode(this, introduction_point, 'introductionpoint');
 
-    handshake_buffer.write(static_cast<uint8_t>(2));
+    handshake_bytes.writeUInt8(2, 0);
-    handshake_buffer.write(swap_endianness(introducee->get_ip_address().to_int()));
+    handshake_bytes.writeInt32BE(parseIp(introducee.ip), 1);
-    handshake_buffer.write(introducee->get_or_port());
+    handshake_bytes.writeUInt16BE(introducee.or_port, 5);
-    handshake_buffer.write(introducee->get_identity_fingerprint());
+    introducee.identity_fingerprint.copy(handshake_bytes, 7);
-    handshake_buffer.write(static_cast<payload_size_type>(introducee->get_onion_key().get_size()));
+    handshake_bytes.writeUInt16BE(introducee.onion_key.length, 27);
-    handshake_buffer.write(introducee->get_onion_key());
+    introducee.onion_key.copy(handshake_bytes, 29);
-    handshake_buffer.write(rendezvous_cookie);
+    rendezvous_cookie.copy(handshake_bytes, 29 + introducee.onion_key.length);
-    handshake_buffer.write(rendezvous_circuit->_extend_node->get_key_agreement().get_public_key());
+    rendezvous_circuit._extend_node.key_agreement.public_key.copy(handshake_bytes, 29 + introducee.onion_key.length + rendezvous_cookie.length);
 
-    const handshake_encrypted = hybrid_encryption::encrypt(
+    const handshake_encrypted = hybrid_encrypt(handshake_bytes, introduction_point.service_key);
-      handshake_bytes,
-      introduction_point.service_key);
 
       //
       // compose the final payload.
@@ -722,11 +828,11 @@ class Circuit {
 
     if (await this.wait_for_state(States.rendezvous_introduced))
     {
-      this._logger.debug("circuit::rendezvous_introduce() [or: %s, state: introduced]", introduction_point.name);
+      this._logger.debug("Circuit.rendezvous_introduce() [or: %s, state: introduced]", introduction_point.name);
     }
     else
     {
-      this._logger.error("circuit::rendezvous_introduce() [or: %s, is_rendezvous_introduced() == false]", introduction_point.name);
+      this._logger.error("Circuit.rendezvous_introduce() [or: %s, is_rendezvous_introduced() == false]", introduction_point.name);
 
       //
       // we cannot expect the rendezvous will be completed.
@@ -737,28 +843,40 @@ class Circuit {
 
     if (await rendezvous_circuit.wait_for_state(States.rendezvous_completed))
     {
-      this._logger.debug("circuit::rendezvous_introduce() [or: %s, state: completed]", introduction_point.name);
+      this._logger.debug("Circuit.rendezvous_introduce() [or: %s, state: completed]", introduction_point.name);
     }
     else
     {
-      this._logger.error("circuit::rendezvous_introduce() [or: %s, is_rendezvous_completed() == false]", introduction_point.name);
+      this._logger.error("Circuit.rendezvous_introduce() [or: %s, is_rendezvous_completed() == false]", introduction_point.name);
     }
   }
 
+  /**
+   * @returns {boolean}
+   */
   is_ready() {
-    return this.state === States.ready;
+    return this._state === States.ready;
   }
 
+  /**
+   * @returns {boolean}
+   */
   is_rendezvous_established() {
-    return this.state === States.rendezvous_established;
+    return this._state === States.rendezvous_established;
   }
 
+  /**
+   * @returns {boolean}
+   */
   is_rendezvous_completed() {
-    return this.state === States.rendezvous_completed;
+    return this._state === States.rendezvous_completed;
   }
 
+  /**
+   * @returns {boolean}
+   */
   is_rendezvous_introduced() {
-    return this.state === States.rendezvous_introduced;
+    return this._state === States.rendezvous_introduced;
   }
 }
 

src/tor/CircuitNode.js

@@ -14,6 +14,9 @@ class CircuitNode {
   get onion_router() {
     return this._onion_router;
   }
+  get key_agreement() {
+    return this._handshake;
+  }
 
   constructor (circuit, or, type) {
     this._circuit = circuit;

src/tor/Consensus.js

@@ -13,7 +13,7 @@ function authority_onion_router(name, ip, or_port, dir_port) {
   }
 }
 const authorities = [
-  authority_onion_router("dizum",      "194.109.206.212", 443,  80),
+  //authority_onion_router("dizum",      "194.109.206.212", 443,  80),
   authority_onion_router("Serge",      "66.111.2.131",    9001, 9030),
   authority_onion_router("moria1",     "128.31.0.34",     9101, 9131),
   //authority_onion_router("tor26",      "86.59.21.38",     443,  80},
@@ -41,16 +41,28 @@ const router_status_flags = [
 ];
 
 class Consensus {
+  /** @type {number} */
   _allowed_dir_flags = OR.fast | OR.valid | OR.running | OR.v2dir;
+  /** @type {number[]} */
   _allowed_dir_ports = [];
-  _max_try_count = 3;
+  /** @type {number} */
+  _max_try_count = 5;
+  /** @type {Object<string,OnionRouter>} */
   _onion_router_map = {};
+  /** @type {number} */
   _valid_until = 0;
 
+  /**
+   * @param {Logger} logger
+   */
   constructor (logger) {
     this._logger = logger;
   }
 
+  /**
+   * @param {string} [cached_consensus_path]
+   * @returns {Promise<void>}
+   */
   async fetch(cached_consensus_path) {
     let have_valid_consensus = false;
     let force_download = false;
@@ -102,10 +114,17 @@ class Consensus {
     }
   }
 
+  /**
+   * @param {number} ports
+   */
   set_allowed_dir_ports(...ports) {
     this._allowed_dir_ports = [...ports];
   }
 
+  /**
+   * @param {{dir_ports:number[],or_ports:number[],flags:number,forbidden_routers:string[]}} criteria
+   * @returns {OnionRouter[]}
+   */
   get_onion_routers_by_criteria(criteria) {
     const candidates = [];
     for (const key in this._onion_router_map) {
@@ -130,6 +149,10 @@ class Consensus {
     return candidates;
   }
 
+  /**
+   * @param {{dir_ports:number[],or_ports:number[],flags:number,forbidden_routers:string[]}} criteria
+   * @returns {OnionRouter}
+   */
   get_random_onion_router_by_criteria(criteria) {
     const candidates = this.get_onion_routers_by_criteria(criteria);
     if (candidates.length === 0) {

src/tor/HiddenServiceConnector.js

@@ -14,13 +14,15 @@ class HiddenServiceConnector {
   _responsible_directory_list = [];
   /** @type Buffer */
   _rendezvous_cookie = null;
+  _logger;
 
-  constructor (rendezvous_circuit, onion) {
+  constructor (rendezvous_circuit, onion, logger) {
     this._rendezvous_circuit = rendezvous_circuit;
     this._socket = rendezvous_circuit.tor_socket;
     this._consensus = rendezvous_circuit.tor_socket.onion_router.consensus;
     this._onion = onion;
     this._permanent_id = Buffer.from(base32.decode(onion));
+    this._logger = logger;
   }
 
   async connect() {
@@ -112,8 +114,8 @@ class HiddenServiceConnector {
     const time_period = (time() + (permanent_id_byte * 86400 / 256)) / 86400;
 
     const secret_bytes = Buffer.alloc(5);
-    secret_bytes.writeInt32BE(time_period);
+    secret_bytes.writeInt32BE(time_period, 0);
-    secret_bytes.writeUInt8(replica);
+    secret_bytes.writeUInt8(replica, 4);
 
     return sha1(secret_bytes);
   }
@@ -189,7 +191,7 @@ class HiddenServiceConnector {
       //
       // create the directory stream on the directory circuit.
       //
-      const directory_stream = directory_circuit.create_dir_stream();
+      const directory_stream = await directory_circuit.create_dir_stream();
 
       if (!directory_stream)
       {
@@ -308,7 +310,7 @@ class HiddenServiceConnector {
       this._logger.info("\tConnected...");
 
       this._logger.info(
-        "\tExtending circuit to introduction point '%s' (%s:%u)",
+        "\tExtending circuit to introduction point '%s' (%s:%i)",
         introduction_point.name,
         introduction_point.ip,
         introduction_point.or_port);

src/tor/OnionRouter.js

@@ -23,8 +23,12 @@ class OnionRouter {
   /** @type Buffer */
   _identity_fingerprint = null;
   _descriptor_fetched = false;
+  /** @type Buffer */
   _ntor_onion_key = null;
+  /** @type Buffer */
   _service_key = null;
+  /** @type Buffer */
+  _onion_key = null;
 
   constructor (consensus, nickname, ip, or_port, dir_port, identity_fingerprint) {
     this._consensus = consensus;
@@ -49,6 +53,7 @@ class OnionRouter {
   get identity_fingerprint() { return this._identity_fingerprint; }
   set service_key(key) { this._service_key = key; }
   get service_key() { return this._service_key; }
+  get onion_key() { return this._onion_key; }
 
   async fetch_descriptor() {
     const descriptor = await this._consensus.get_onion_router_descriptor(this._identity_fingerprint);

src/utils/crypto.js

@@ -17,3 +17,62 @@ function sha1(data) {
   return h.update(data).digest();
 }
 exports.sha1 = sha1;
+
+function rsa_1024(public_key) {
+  const rsa = crypto.createPublicKey(public_key);
+  return {
+    encrypt(data) {
+      return crypto.publicEncrypt(rsa, data);
+    }
+  }
+}
+
+function aes_ctr_128(key) {
+  const aes = crypto.createCipheriv('aes-128-ctr', key, Buffer.alloc(16));
+  return {
+    encrypt(data) {
+      return aes.update(data);
+    }
+  }
+}
+
+const KEY_LEN = 16;
+const PK_ENC_LEN = 128;
+const PK_PAD_LEN = 42;
+const PK_DATA_LEN = PK_ENC_LEN - PK_PAD_LEN;
+const PK_DATA_LEN_WITH_KEY = PK_DATA_LEN - KEY_LEN;
+
+/**
+ * @param {Buffer} data
+ * @param {Buffer} public_key
+ * @returns {Buffer}
+ */
+function hybrid_encrypt(data, public_key) {
+  if (data.length < PK_DATA_LEN)
+  {
+    return rsa_1024(public_key).encrypt(data);
+  }
+
+  const random_key = crypto.randomBytes(KEY_LEN);
+
+  //
+  // RSA( K | M1 ) --> C1
+  //
+  const k_and_m1 = Buffer.concat([
+    random_key, data.slice(0, PK_DATA_LEN_WITH_KEY)
+  ], random_key.length + PK_DATA_LEN_WITH_KEY);
+
+  const c1 = rsa_1024(public_key).encrypt(k_and_m1);
+
+  //
+  // AES_CTR(M2)  --> C2
+  //
+  const m2 = data.slice(PK_DATA_LEN_WITH_KEY);
+  const c2 = aes_ctr_128(random_key).encrypt(m2);
+
+  //
+  // C1 | C2
+  //
+  return  Buffer.concat([ c1, c2 ], c1.length + c2.length);
+}
+exports.hybrid_encrypt = hybrid_encrypt;