Initial.

requirements.txt

@@ -0,0 +1,5 @@
+cbor2
+pyzmq
+pyvis
+aiohttp
+pycryptodome

yafn-tracker/.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.bak
+

yafn-tracker/__main__.py

@@ -0,0 +1,168 @@
+import time
+import socket
+import asyncio
+import ipaddress
+
+import cbor2
+from aiohttp import web
+from Crypto.Hash import SHA256, RIPEMD160
+from Crypto.PublicKey import RSA
+
+def ripemd160(data):
+  hash = RIPEMD160.new()
+  hash.update(data)
+
+  return hash.digest()
+
+def sha256(data):
+  hash = SHA256.new()
+  hash.update(data)
+
+  return hash.digest()
+
+def generate_uid(pubkey):
+  pubkey = RSA.import_key(pubkey)
+  pubkey = pubkey.public_key()
+
+  n = pubkey.n.to_bytes(128, 'big')
+  e = pubkey.e.to_bytes(3, 'big')
+
+  return ripemd160(sha256(n + e))
+
+async def _readall(reader, size):
+  buffer = b''
+
+  while len(buffer) < size:
+    buffer += await reader.read(size - len(buffer))
+
+  return buffer
+
+class Peer:
+  def __init__(self, addr):
+    self.addr = addr
+    self.uid = None
+    self.last_check = 0
+    self.failed_checks = 0
+    self.works = False
+    self.latency = 0
+
+  async def check(self):
+    self.last_check = time.time()
+
+    try:
+      addr = socket.gethostbyname(self.addr)
+      addr = ipaddress.ip_address(addr)
+      if not addr.is_global:
+        raise Exception
+
+      reader, writer = await asyncio.wait_for(
+        asyncio.open_connection(str(addr), 49871),
+        timeout=5
+      )
+
+      data = await asyncio.wait_for(
+        _readall(reader, 172),
+        timeout=5
+      )
+
+      if len(data) != 172:
+        raise Exception
+
+      if data[:10] != b'YAFN HELLO':
+        raise Exception
+
+      self.uid = generate_uid(data[10:])
+
+      writer.write(b'YAFN TRACKER CHECK' + b'\x00' * 154)
+      await asyncio.wait_for(
+        writer.drain(),
+        timeout=5
+      )
+    except:
+      self.failed_checks += 1
+      self.works = False
+
+      return False
+    finally:
+      try:
+        writer.close()
+        await asyncio.wait_for(
+          writer.wait_closed(),
+          timeout=2
+        )
+      except:
+        pass
+
+    self.failed_checks = 0
+    self.works = True
+    self.latency = time.time() - self.last_check
+
+    return True
+
+class Tracker:
+  def __init__(self):
+    self.peers = []
+
+  async def add(self, addr):
+    for peer in self.peers:
+      if peer.addr == addr:
+        return await peer.check()
+
+    peer = Peer(addr)
+    if await peer.check():
+      self.peers.append(peer)
+
+      return True
+
+    return False
+
+  async def watch(self):
+    while True:
+      to_delete = []
+
+      for peer in self.peers:
+        if not await peer.check():
+          if peer.failed_checks >= 3:
+            to_delete.append(peer)
+
+      for peer in to_delete:
+        self.peers.remove(peer)
+
+      await asyncio.sleep(60)
+
+tracker = Tracker()
+
+async def request_handler(request):
+  remote_addr = request.headers.get('YAFN-Remote-Address', request.remote)
+
+  is_accessible = await tracker.add(remote_addr)
+  peers = tracker.peers.copy()
+  peers = [
+    {
+      'address': peer.addr,
+      'uid': peer.uid,
+      'latency': int(peer.latency),
+      'last_check': int(peer.last_check)
+    } for peer in peers if peer.works and peer.addr != remote_addr
+  ]
+  peers.sort(key=lambda peer: (peer['last_check'], peer['latency']))
+
+  return web.Response(
+    body=cbor2.dumps({
+      'remote_addr': remote_addr,
+      'is_accessible': is_accessible,
+      'peers': peers
+    })
+  )
+
+async def main():
+  asyncio.ensure_future(tracker.watch())
+
+  app = web.Application()
+  app.add_routes([
+    web.get('/track', request_handler),
+  ])
+
+  return app
+
+web.run_app(main(), port=49873)

yafn/.gitignore

@@ -0,0 +1,3 @@
+__pycache__/
+*.bak
+

yafn/__main__.py

@@ -0,0 +1,310 @@
+import os
+import os.path
+import sys
+import math
+import time
+import atexit
+import struct
+import argparse
+
+from pathlib import Path
+
+from . import log
+from . import yafn
+
+from pyvis.network import Network
+
+class Progress:
+  def __init__(self, max):
+    self.max = max
+
+    self.CHARS = ('/', '-', '\\', '|')
+
+    self.index = 0
+    self.length = 0
+    self.ready = 0
+
+    self._display(0)
+
+  def _render(self):
+    text = f'{self.ready}/{self.max} piece{"s" if self.ready != 1 else ""} ready ... {self.CHARS[self.index]}\n'
+    
+    self.index += 1
+
+    if self.index == len(self.CHARS):
+      self.index = 0
+
+    self.length = len(text)
+
+    return text
+
+  def _display(self, ready):
+    text = self._render()
+
+    sys.stdout.write(text)
+    sys.stdout.flush() 
+
+  def update(self, ready):
+    self.ready += ready
+
+    sys.stdout.write("\033[F")
+    sys.stdout.write("\033[K")
+    sys.stdout.write(self._render())
+    sys.stdout.flush() 
+
+class Metafile:
+  HEADER = b'\x80YAFN-METAFILE\x00\x00'
+  HEADER_LEN = len(HEADER)
+
+  def __init__(self, filename, size, pieces):
+    self.filename = filename
+    self.size = size
+    self.pieces = pieces
+
+  def save(self, path):
+    with open(path, 'wb') as f:
+      f.write(Metafile.HEADER)
+
+      filename = self.filename.encode('UTF-8')
+      filename_len = len(filename)
+  
+      f.write(struct.pack('!I', filename_len))
+      f.write(filename)
+
+      f.write(struct.pack('!L', self.size))
+
+      pieces_count = len(self.pieces)
+
+      f.write(struct.pack('!L', pieces_count))
+    
+      for hash in self.pieces:
+        f.write(hash)
+
+  @staticmethod
+  def load(path):
+    with open(path, 'rb') as f:
+      header = f.read(Metafile.HEADER_LEN)
+      if header != Metafile.HEADER:
+        raise ValueError
+
+      filename_len = f.read(4)
+      filename_len = struct.unpack('!I', filename_len)[0]
+
+      filename = f.read(filename_len)
+      filename = filename.decode('UTF-8')
+      filename = Path(filename).name
+
+      size = f.read(4)
+      size = struct.unpack('!I', size)[0]
+
+      pieces_count = f.read(4)
+      pieces_count = struct.unpack('!L', pieces_count)[0]
+
+      if pieces_count < 1:
+        raise ValueError
+
+      pieces = []
+
+      while pieces_count > 0:
+        piece_hash = f.read(76)
+        if len(piece_hash) != 76:
+          raise ValueError
+
+        pieces.append(piece_hash)
+
+        pieces_count -= 1
+
+    return Metafile(filename, size, pieces)
+
+COLORS = [
+  (0,   255, 0),
+  (255, 255, 0),
+  (255, 0,   0)
+]
+
+def pick_color(index):
+  index = min(index, 6) * 0.2
+  n3 = 0
+
+  if index <= 0:
+    n1 = 0
+    n2 = 0
+  elif index >= 1:
+    n1 = len(COLORS) - 1
+    n2 = len(COLORS) - 1
+  else:
+    index = index * (len(COLORS) - 1)
+    n1 = math.floor(index)
+    n2 = n1 + 1
+    n3 = index - n1
+
+  color = (
+    (COLORS[n2][0] - COLORS[n1][0]) * n3 + COLORS[n1][0],
+    (COLORS[n2][1] - COLORS[n1][1]) * n3 + COLORS[n1][1],
+    (COLORS[n2][2] - COLORS[n1][2]) * n3 + COLORS[n1][2]
+  )
+
+  return f'#{int(color[0]):02x}{int(color[1]):02x}{int(color[2]):02x}'
+
+def build_graph(graph, map, index=0):
+  uid = yafn.encode_uid(map.uid)
+
+  graph.add_node(
+    uid,
+    label=f'{uid[:6]}{" (this peer)" if index == 0 else ""}',
+    title=uid,
+    color=pick_color(index)
+  )
+
+  for submap in map.submaps:
+    subuid = build_graph(graph, submap, index + 1)
+    
+    graph.add_edge(uid, subuid)
+
+  return uid
+
+parser = argparse.ArgumentParser()
+parser.add_argument(
+  '-S', '--start',
+  help='Start up a peer.',
+  action='store_true'
+)
+
+parser.add_argument(
+  '-a', '--address',
+  help='Set a custom external address.',
+  type=str
+)
+
+parser.add_argument(
+  '-c', '--crawl',
+  help='Create a map of the network.',
+  action='store_true'
+)
+
+parser.add_argument(
+  '-s', '--share',
+  help='Share a file to the network',
+  action='append',
+  type=str,
+  metavar='PATH'
+)
+
+parser.add_argument(
+  '-q', '--query',
+  help='Query a file from the network',
+  action='append',
+  type=str,
+  metavar='METAFILE'
+)
+
+args = parser.parse_args()
+
+if args.crawl or args.share or args.query:
+  interface = yafn.Interface.connect()
+
+  atexit.register(interface.close)
+
+if args.crawl:
+  log.info(f'Building a map of the network...')
+
+  map = interface.crawl()
+  if not map:
+    log.fatal('Failed to crawl the network.')
+ 
+  network = Network(
+    height='100%',
+    width='100%',
+    bgcolor='#222222',
+    font_color='white'
+  )
+
+  build_graph(network, map)
+
+  adj_list = network.get_adj_list()
+  for node in network.nodes:
+    node['value'] = min(max(len(adj_list[node['id']]), 1), 10)
+
+  filename = f'map_{int(time.time())}.html'
+  network.save_graph(filename)
+
+  log.info(f'Network map is saved as \'{filename}\'.')
+  
+if args.share:
+  for path in args.share:
+    if not os.path.isfile(path):
+      log.fatal(f'Not a valid file: \'{path}\'.')
+
+    log.info(f'Share \'{path}\'.')
+
+    with open(path, 'rb') as f:
+      progress = Progress('_')
+      pieces = []
+      size = 0
+
+      while True:
+        piece = f.read(yafn.Piece.PIECE_SIZE)
+        if not piece:
+          break
+
+        hash = interface.save(piece)
+        if not hash:
+          log.error('Failed to save a piece.')
+
+          continue
+
+        pieces.append(hash)
+        progress.update(1)
+
+        size += len(piece)
+
+    filename = os.path.basename(path)
+    metafile_name = f'{filename}.ynmf'
+
+    metafile = Metafile(
+      filename,
+      size,
+      pieces
+    )
+
+    try:
+      metafile.save(metafile_name)
+    except:
+      log.fatal('Failed to save the metafile.')
+
+    log.info(f'Metafile is saved as \'{metafile_name}\'.')
+
+    log.info('Announcing the neighbour peers...')
+    
+    if interface.announce():
+      log.info('Done.')
+    else:
+      log.error('Announcing failed.')
+ 
+if args.query:
+  for path in args.query:
+    try:
+      metafile = Metafile.load(path)
+    except:
+      log.fatal(f'Cannot open metafile: \'{path}\'.')
+
+    log.info(f'Query \'{metafile.filename}\'.')
+
+    with open(metafile.filename, 'wb') as f:
+      progress = Progress(len(metafile.pieces))
+
+      for hash in metafile.pieces:
+        piece = interface.query(hash)
+        if not piece:
+          log.error(f'Piece {hash[:32].hex()} is not available.')
+
+          continue
+
+        f.write(piece)
+        progress.update(1)
+
+    log.info(f'File is saved as \'{metafile.filename}\'.')
+
+if args.start:
+  peer = yafn.Peer()
+  peer.start(remote_addr=args.address)

yafn/log.py

@@ -0,0 +1,18 @@
+import sys
+import logging
+
+logging.basicConfig(format="[%(asctime)s] %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", level=logging.INFO)
+
+def info(message):
+  logging.info(message)
+
+def warning(message):
+  logging.warning(message)
+
+def error(message):
+  logging.error(message)
+
+def fatal(message):
+  logging.error(message)
+
+  sys.exit(1)

yafn/yafn.py

@@ -0,0 +1,1504 @@
+import os
+import os.path
+import sys
+import uuid
+import time
+import zlib
+import random
+import struct
+import socket
+import pathlib
+import threading
+import urllib.request
+
+import zmq
+import cbor2
+
+from Crypto.Hash import SHA256, RIPEMD160
+from Crypto.Cipher import AES, Salsa20, PKCS1_OAEP
+from Crypto.Random import get_random_bytes
+from Crypto.PublicKey import RSA
+
+from . import log
+
+ALPHABET = '286USqFxzKsmMBP9c4TOyECkefQZ7otHAjYh5aN1WiLRprnIGwgulV0dDX'
+
+def encode_uid(uid):
+  result = '' 
+
+  n = int.from_bytes(uid, 'big')
+
+  while n >= 58:
+    m = n % 58
+    n //= 58
+    result += ALPHABET[m]
+
+  if n > 0:
+    result += ALPHABET[n]
+
+  return result
+
+class Timer(threading.Thread):
+  def __init__(self, interval, callback, preflight=0):
+    super().__init__()
+    self.interval = interval
+    self.callback = callback
+    self.preflight = preflight
+    self.event = threading.Event()
+
+  def run(self):
+    if self.preflight:
+      time.sleep(self.preflight)
+
+      self.callback()
+
+    while not self.event.wait(self.interval):
+      self.callback()
+
+def spawn_thread(target, *args):
+  thread = threading.Thread(
+    target=target,
+    args=args
+  )
+  thread.daemon = True
+  thread.start()
+
+def adler32(data):
+  return zlib.adler32(data) & 0xffffffff
+
+def ripemd160(data):
+  hash = RIPEMD160.new()
+  hash.update(data)
+
+  return hash.digest()
+
+def sha256(data):
+  hash = SHA256.new()
+  hash.update(data)
+
+  return hash.digest()
+
+def generate_uid(pubkey):
+  pubkey = pubkey.public_key()
+
+  n = pubkey.n.to_bytes(128, 'big')
+  e = pubkey.e.to_bytes(3, 'big')
+
+  return ripemd160(sha256(n + e))
+
+def RSA_generate_keypair():
+  keypair = RSA.generate(1024)
+  
+  return keypair
+
+def RSA_encrypt(data, key):
+  key = key.public_key()
+  cipher = PKCS1_OAEP.new(key)
+
+  return cipher.encrypt(data)
+
+def RSA_decrypt(data, key):
+  cipher = PKCS1_OAEP.new(key)
+
+  return cipher.decrypt(data)
+
+def AES_encrypt(data, key):
+  cipher = AES.new(key, AES.MODE_GCM)
+  nonce = cipher.nonce
+
+  data, tag = cipher.encrypt_and_digest(data)
+  
+  return (data, nonce, tag)
+
+def AES_decrypt(data, key, nonce, tag):
+  cipher = AES.new(key, AES.MODE_GCM, nonce=nonce)
+
+  data = cipher.decrypt(data)
+  
+  cipher.verify(tag)
+
+  return data
+
+def RSA_AES_hybrid_encrypt(data, public_key):
+  public_key = public_key.public_key()
+  cipher = PKCS1_OAEP.new(public_key)
+
+  key = get_random_bytes(32)
+  data, nonce, tag = AES_encrypt(data, key)
+
+  key = key + nonce + tag
+  key = cipher.encrypt(key)
+
+  return (data, key)
+
+def RSA_AES_hybrid_decrypt(data, key, private_key):
+  cipher = PKCS1_OAEP.new(private_key)
+  
+  key = cipher.decrypt(key)
+  if len(key) != 64:
+    raise ValueError
+
+  key, nonce, tag = key[:32], key[32:48], key[48:64]
+
+  return AES_decrypt(data, key, nonce, tag)
+
+def salsa20_create_encryptor():
+  key = get_random_bytes(32)
+  cipher = Salsa20.new(key=key)
+
+  return (cipher.nonce + key, cipher)
+
+def salsa20_create_decryptor(key):
+  if len(key) != 40:
+    raise ValueError
+
+  cipher = Salsa20.new(key=key[8:], nonce=key[:8])
+  
+  return cipher
+
+def chunked(data, size):
+  return [data[i:i+size] for i in range(0, len(data), size)]
+
+class MessageKind:
+  PING      = b'P'
+  PONG      = b'O'
+  BYE       = b'B'
+
+  QUERY     = b'Q'
+  QUERY_HIT = b'H'
+  NOTAVAIL  = b'N'
+
+  CRAWL     = b'C'
+  MAP       = b'M'
+
+  ANNOUNCE  = b'A'
+
+class Message:
+  def __init__(self, kind, uid=None, **fields):
+    self.kind = kind
+    self.uid = uid if uid else get_random_bytes(16)
+
+    self.__dict__.update(**fields)
+
+    self._timestamp = time.time()
+
+  @property
+  def age(self):
+    return time.time() - self._timestamp
+
+class Part:
+  def __init__(self, data, size, n, checksum):
+    self.data = data
+    self.size = size
+    self.n = n
+    self.checksum = checksum
+
+class Piece:
+  HEADER      = b'\x80YAFN-PIECE\x00\x00'
+  HEADER_SIZE = len(HEADER)
+  PIECE_SIZE  = 512 * 1024
+  PART_SIZE   = 1024
+
+  def __init__(self, timestamp, hash, parts, parts_count):
+    self.timestamp = timestamp
+    self.hash = hash
+    self.parts = parts
+    self.parts_count = parts_count
+
+  def join(self):
+    data = b''
+
+    for part in self.parts:
+      data += part.data
+
+    return data
+
+  def dump(self, fd):
+    fd.write(Piece.HEADER)
+    
+    fd.write(self.hash)
+    fd.write(struct.pack('!Q', int(self.timestamp)))
+    fd.write(struct.pack('!H', self.parts_count))
+
+    for order, part in zip(range(self.parts_count), self.parts):
+      fd.write(struct.pack('!H', order))
+      fd.write(struct.pack('!L', part.checksum))
+      fd.write(struct.pack('!H', part.size))
+
+      fd.write(part.data)
+
+  @staticmethod
+  def load(fd):
+    header = fd.read(Piece.HEADER_SIZE)
+    if header != Piece.HEADER:
+      raise ValueError
+
+    hash = fd.read(32)
+    if len(hash) != 32:
+      raise ValueError
+
+    timestamp = fd.read(8)
+    timestamp = struct.unpack('!Q', timestamp)[0]
+
+    parts_count = fd.read(2)
+    parts_count = struct.unpack('!H', parts_count)[0]
+    if parts_count < 1:
+      raise ValueError
+
+    actual_hash = SHA256.new()
+    parts = []
+
+    for n in range(parts_count):
+      order = fd.read(2)
+      order = struct.unpack('!H', order)[0]
+      if order != n:
+        raise ValueError
+
+      checksum = fd.read(4)
+      checksum = struct.unpack('!L', checksum)[0]
+
+      size = fd.read(2)
+      size = struct.unpack('!H', size)[0]
+      if size < 1:
+        raise ValueError
+
+      data = fd.read(size)
+      if len(data) != size:
+        raise ValueError
+
+      if adler32(data) != checksum:
+        raise ValueError
+
+      parts.append(Part(
+        data,
+        size,
+        order,
+        checksum
+      ))
+
+      actual_hash.update(data)
+
+    if len(parts) != parts_count:
+      raise ValueError
+
+    if actual_hash.digest() != hash:
+      raise ValueError
+
+    return Piece(
+      timestamp,
+      hash,
+      parts,
+      parts_count
+    )
+
+  @staticmethod
+  def create(data):
+    hash = SHA256.new()
+    hash.update(data)
+
+    parts = chunked(data, Piece.PART_SIZE)
+
+    return Piece(
+      time.time(),
+      hash.digest(),
+      [
+        Part(
+          part,
+          len(part),
+          n,
+          adler32(part)
+        ) for n, part in zip(range(len(parts)), parts)
+      ],
+      len(parts)
+    )
+
+class Storage:
+  YAFN_DIR = os.path.join(
+    pathlib.Path.home(),
+    'yafn'
+  )
+
+  STORAGE_DIR = os.path.join(
+    YAFN_DIR,
+    '.storage'
+  )
+
+  KEYPAIR_FILE = os.path.join(
+    YAFN_DIR,
+    'keypair.pem',
+  )
+
+  TRACKERS_FILE = os.path.join(
+    YAFN_DIR,
+    'trackers.txt'
+  )
+
+  @staticmethod
+  def setup():
+    if not os.path.isdir(Storage.YAFN_DIR):
+      os.mkdir(Storage.YAFN_DIR)
+
+    if not os.path.isdir(Storage.STORAGE_DIR):
+      os.mkdir(Storage.STORAGE_DIR)
+
+    if not os.path.isfile(Storage.TRACKERS_FILE):
+      open(Storage.TRACKERS_FILE, 'w').close()
+    
+  @staticmethod
+  def get_keypair():
+    if not os.path.isfile(Storage.KEYPAIR_FILE):
+      keypair = RSA_generate_keypair()
+
+      with open(Storage.KEYPAIR_FILE, 'wb') as f:
+        f.write(keypair.export_key())
+
+      return keypair
+
+    with open(Storage.KEYPAIR_FILE, 'rb') as f:
+      keypair = RSA.import_key(f.read())
+
+    return keypair
+
+  @staticmethod
+  def get_trackers():
+    if not os.path.isfile(Storage.TRACKERS_FILE):
+      return []
+
+    with open(Storage.TRACKERS_FILE, 'r') as f:
+      lines = f.readlines()
+      lines = map(lambda line: line.strip(), lines)
+      lines = filter(bool, lines)
+
+      return list(set(lines))
+
+  @staticmethod
+  def find_piece(hash):
+    path = os.path.join(Storage.STORAGE_DIR, hash.hex())
+    if not os.path.isfile(path):
+      return None
+
+    with open(path, 'rb') as f:
+      try:
+        piece = Piece.load(f)
+        if piece.hash != hash:
+          os.remove(path)
+
+          return None
+
+        return piece
+      except:
+        return None
+
+  @staticmethod
+  def save_piece(piece):
+    if type(piece) is not Piece:
+      piece = Piece.create(piece)
+
+    path = os.path.join(Storage.STORAGE_DIR, piece.hash.hex())
+    
+    with open(path, 'wb') as f:
+      piece.dump(f)
+
+    return piece
+
+  @staticmethod
+  def list_pieces():
+    pieces = set()
+    files = os.listdir(Storage.STORAGE_DIR)
+
+    for file in files:
+      piece = Storage.find_piece(bytes.fromhex(file))
+      pieces.add(piece.hash)
+
+    return pieces
+
+class YAFNError(Exception): pass
+
+class CachedMessage:
+  def __init__(self, kind, uid):
+    self.kind = kind
+    self.uid = uid
+
+    self._timestamp = time.time()
+
+  @property
+  def age(self):
+    return time.time() - self._timestamp
+
+class Map:
+  def __init__(self, uid, submaps):
+    self.uid = uid
+    self.submaps = submaps
+    self.submaps_count = len(submaps)
+
+  def dump(self):
+    data = b''
+
+    data += self.uid
+    data += struct.pack('!H', self.submaps_count)
+
+    for submap in self.submaps:
+      data += submap.dump()
+
+    return data
+
+  def split(self):
+    data = self.dump()
+
+    return chunked(data, Piece.PART_SIZE)
+
+  @staticmethod
+  def _drain(data):
+    if len(data) < 20 + 2:
+      raise ValueError
+
+    uid = data[:20]
+    submaps_count = struct.unpack('!H', data[20:22])[0]
+
+    data = data[22:]
+
+    submaps = []
+    while submaps_count > 0:
+      data, submap = Map._drain(data)
+      submaps.append(submap)
+
+      submaps_count -= 1
+
+    return data, Map(uid, submaps)
+
+  @staticmethod
+  def create(data):
+    data, map = Map._drain(data)
+
+    if len(data) > 0:
+      raise ValueError
+
+    return map
+
+class Connection:
+  def __init__(self, peer, socket, addr, is_inbound=True, reconnect_attempts=0):
+    self._peer = peer
+    self._socket = socket
+    self.addr = addr
+    self.is_inbound = is_inbound
+    self._reconnect_attempts = reconnect_attempts
+
+    self.uid = None
+    self.is_alive = True
+    self.near_pieces = set()
+    self._near_pieces_purge_timestamp = time.time()
+    self._send_lock = threading.Lock()
+    self._dont_reconnect = False
+    self._timestamp = time.time()
+    self._queries_pending = 0
+    self._messages_pending = 0
+    self._queue = []
+    self._cache = []    
+
+    self._remote_pubkey = None    
+
+    self._watchdog = Timer(30 if self.is_inbound else 60, self._watch)
+    self._watchdog.start()
+
+    self._announcer = Timer(60*10, self.announce)
+    self._announcer.start()
+
+  def _watch(self):
+    if not self.is_alive:
+      return
+
+    if not self._remote_pubkey:
+      self.close()
+
+      return
+
+    attempts = 0
+    while True:
+      attempts += 1
+
+      if attempts > 3:
+        self.close()
+
+        return
+
+      message = Message(MessageKind.PING)
+
+      try:
+        self.send(message)
+      except:
+        continue
+
+      response = self.wait(
+        lambda m: m.uid == message.uid and m.kind == MessageKind.PONG,
+        timeout=15
+      )
+      if not response:
+        continue
+
+      break
+
+    self._cache = [
+      message for message in self._cache if message.age < 60*60
+    ]
+
+    self._queue = [
+      message for message in self._queue if message.age < 60*5
+    ]
+
+    if time.time() - self._near_pieces_purge_timestamp > 60*60*8:
+      self.near_pieces = set()
+
+      self._near_pieces_purge_timestamp = time.time()
+
+  def _recvall(self, size):
+    buffer = b''
+
+    while len(buffer) != size:
+      buffer += self._socket.recv(size - len(buffer))
+
+    return buffer
+
+  def _sendall(self, data):
+    self._socket.sendall(data)
+
+  @property
+  def age(self):
+    return time.time() - self._timestamp
+
+  @property
+  def is_ok(self):
+    if self.is_alive:
+      return True
+
+    try:
+      self._peer.connections.remove(self)
+    except:
+      pass
+
+    return False
+
+  def is_cached(self, message):
+    for other_message in self._cache:
+      if message.uid == other_message.uid:
+        return True
+
+    return False
+
+  def cache(self, message):
+    if not self.is_cached(message):
+      self._cache.append(CachedMessage(message.kind, message.uid))
+
+  def wait(self, tester, timeout=60):
+    start_ts = time.time()
+
+    while time.time() - start_ts < timeout:
+      queue = self._queue.copy()
+      for message in queue:
+        if tester(message):
+          try:
+            self._queue.remove(message)
+          except:
+            pass
+
+          return message
+
+      if not self.is_alive:
+        return None
+
+      time.sleep(1)
+
+  def close(self):
+    if not self.is_alive:
+      return
+
+    try:
+      self.send(Message(MessageKind.BYE))
+    except:
+      pass
+    finally:
+      try:
+        self._socket.close()
+      except:
+        pass
+
+      self._watchdog.event.set()
+      self._announcer.event.set()
+
+      try:
+        self._peer.connections.remove(self)
+      except:
+        pass
+
+      self.is_alive = False
+
+      if not self.is_inbound:
+        log.warning(f'`{self.addr}` ({self.encoded_uid if self.uid else "n/a"}): Connection lost.')
+
+        if self._dont_reconnect or self._reconnect_attempts >= 5:
+          return
+
+        self._reconnect_attempts += 1
+
+        time.sleep(10 * self._reconnect_attempts)
+
+        self._peer.connect_to(self.addr, self._reconnect_attempts)
+
+  def query(self, hash, mid, ttl=7): 
+    self._queries_pending += 1
+
+    try:
+      self.send(
+        Message(
+          MessageKind.QUERY,
+          uid=mid,
+          hash=hash,
+          ttl=ttl
+        )
+      )
+
+      response = self.wait(
+        lambda m: m.uid == mid and m.kind in (MessageKind.QUERY_HIT, MessageKind.NOTAVAIL),
+        timeout=60*8
+      )
+      if response and response.kind == MessageKind.QUERY_HIT:
+        if response.piece.hash != hash:
+          return None
+
+        return response.piece
+    except:
+      pass
+    finally:
+      self._queries_pending -= 1
+
+  def crawl(self, mid, ttl=7): 
+    try:
+      self.send(
+        Message(
+          MessageKind.CRAWL,
+          uid=mid,
+          ttl=ttl
+        )
+      )
+
+      response = self.wait(
+        lambda m: m.uid == mid and m.kind in (MessageKind.MAP, MessageKind.NOTAVAIL), 
+        timeout=60*10
+      )
+      if response and response.kind == MessageKind.MAP:
+        return response.map
+    except:
+      pass
+
+  def announce(self):
+    if not self.is_alive:
+      return
+
+    pieces = Storage.list_pieces()
+    if not pieces:
+      return
+
+    try:
+      self.send(
+        Message(
+          MessageKind.ANNOUNCE,
+          pieces=pieces
+        )
+      )
+    except:
+      pass
+
+  def handshake(self):
+    self._sendall(b'YAFN HELLO' + self._peer.pubkey)
+
+    data = self._recvall(10 + 162)
+
+    head = data[:10]
+    if head != b'YAFN HELLO':
+      raise YAFNError
+      
+    remote_pubkey = RSA.import_key(data[10:]).public_key()
+    remote_uid = generate_uid(remote_pubkey)
+
+    if remote_uid == self._peer.uid or remote_uid in self._peer.connections:
+      self._dont_reconnect = True
+
+      raise YAFNError
+
+    random_data = get_random_bytes(16)
+    data = RSA_encrypt(random_data, remote_pubkey)
+
+    self._sendall(b'CHECK' + data)
+
+    data = self._recvall(5 + 128)
+    
+    head = data[:5]
+    if head != b'CHECK':
+      raise YAFNError
+
+    remote_data = data[5:]
+    remote_data = RSA_decrypt(remote_data, self._peer.keypair)
+
+    self._sendall(b'CHECKED' + remote_data)
+
+    data = self._recvall(7 + 16)
+    
+    head = data[:7]
+    if head != b'CHECKED':
+      raise YAFNError
+
+    if data[7:] != random_data:
+      raise YAFNError
+
+    self._sendall(b'FINISH')
+
+    head = self._recvall(6)
+    if head != b'FINISH':
+      raise YAFNError
+
+    self._remote_pubkey = remote_pubkey
+    self.uid = remote_uid
+    self.encoded_uid = encode_uid(self.uid)
+
+    if not self.is_inbound:
+      log.info(f'`{self.addr}` ({self.encoded_uid}): Connection successful.')
+
+      self._reconnect_attempts = 0
+
+  def _receive_parts(self, key, parts_count):
+    data = b''
+    total = 0
+    cipher = salsa20_create_decryptor(key)
+
+    try:
+      self._socket.settimeout(5)
+
+      while total < parts_count:
+        head = self._recvall(6)
+        
+        checksum = head[:4]
+        checksum = struct.unpack('!L', checksum)[0]
+
+        part_size = head[4:6]
+        part_size = struct.unpack('!H', part_size)[0]
+
+        part = self._recvall(part_size)
+        part = cipher.decrypt(part)
+        if adler32(part) != checksum:
+          raise YAFNError
+
+        data += part
+
+        total += 1
+    finally:
+      self._socket.settimeout(None)
+
+    return data
+
+  def receive(self):
+    head = self._recvall(4 + 2 + 16 + 128)
+
+    checksum = head[:4]
+    size = head[4:6]
+    uid = head[6:22]
+    key = head[22:150]
+
+    checksum = struct.unpack('!L', checksum)[0]
+    size = struct.unpack('!H', size)[0]
+
+    if size > 1024:
+      raise YAFNError
+
+    message = RSA_AES_hybrid_decrypt(message, key, self._peer.keypair)
+    if adler32(message) != checksum:
+      raise YAFNError
+
+    kind = message[:1]
+    payload = message[1:]
+    payload_size = len(payload)
+
+    fields = {}
+
+    if kind in (
+      MessageKind.PING,
+      MessageKind.PONG,
+      MessageKind.BYE,
+      MessageKind.NOTAVAIL
+    ):
+      if payload_size != 0:
+        raise YAFNError
+    elif kind == MessageKind.QUERY:
+      if payload_size != 32 + 1:
+        raise YAFNError
+
+      hash = payload[:32]
+      ttl = payload[32]
+
+      if ttl > 7:
+        raise YAFNError
+
+      fields['hash'] = hash
+      fields['ttl'] = ttl
+    elif kind == MessageKind.QUERY_HIT:
+      if payload_size != 40 + 2:
+        raise YAFNError
+
+      key = payload[:40]
+      parts_count = struct.unpack('!H', payload[40:42])[0]
+
+      data = self._receive_parts(key, parts_count)
+
+      fields['piece'] = Piece.create(data)
+    elif kind == MessageKind.CRAWL:
+      if payload_size != 1:
+        raise YAFNError
+
+      ttl = payload[0]
+
+      if ttl > 7:
+        raise YAFNError
+
+      fields['ttl'] = ttl
+    elif kind == MessageKind.MAP:
+      if payload_size != 40 + 2:
+        raise YAFNError
+
+      key = payload[:40]
+      parts_count = struct.unpack('!H', payload[40:42])[0]
+
+      data = self._receive_parts(key, parts_count)
+
+      fields['map'] = Map.create(data)
+    elif kind == MessageKind.ANNOUNCE:
+      if payload_size != 40 + 4:
+        raise YAFNError
+
+      key = payload[:40]
+      parts_count = struct.unpack('!L', payload[40:44])[0]
+
+      data = self._receive_parts(key, parts_count)
+
+      if len(data) < 32 or len(data) % 32 != 0:
+        raise YAFNError
+
+      fields['pieces'] = set(chunked(data, 32))
+    else:
+      raise YAFNError
+
+    return Message(kind, uid, **fields)
+
+  def _send_parts(self, cipher, parts):
+    for part in parts:
+      checksum = adler32(part)
+      data = cipher.encrypt(part)
+        
+      self._sendall(struct.pack('!L', checksum) + struct.pack('!H', len(data)))
+      self._sendall(data)
+
+  def send(self, message):
+    head = b''
+    payload = message.kind
+
+    if message.kind == MessageKind.QUERY:
+      payload += message.hash
+      payload += bytes([message.ttl])
+    elif message.kind == MessageKind.CRAWL:
+      payload += bytes([message.ttl])
+    elif message.kind in (
+      MessageKind.QUERY_HIT,
+      MessageKind.MAP,
+      MessageKind.ANNOUNCE
+    ):
+      key, cipher = salsa20_create_encryptor()
+
+      if message.kind == MessageKind.MAP:
+        map_parts = message.map.split()
+      elif message.kind == MessageKind.ANNOUNCE:
+        splitted_pieces = b''.join(message.pieces)
+        splitted_pieces = chunked(splitted_pieces, Piece.PART_SIZE)
+
+      payload += key    
+
+      if message.kind == MessageKind.ANNOUNCE:
+        payload += struct.pack('!L', len(splitted_pieces))
+      else:
+        payload += struct.pack('!H', len(map_parts) if message.kind == MessageKind.MAP else message.piece.parts_count)
+   
+    checksum = adler32(payload)
+
+    payload, key = RSA_AES_hybrid_encrypt(payload, self._remote_pubkey)
+
+    head += struct.pack('!L', checksum)
+    head += struct.pack('!H', len(payload))
+    head += message.uid
+    head += key
+
+    try:
+      self._send_lock.acquire()
+
+      self._sendall(head + payload)
+
+      if message.kind == MessageKind.QUERY_HIT:
+        self._send_parts(cipher, [
+          part.data for part in message.piece.parts
+        ])
+      elif message.kind == MessageKind.MAP:
+        self._send_parts(cipher, map_parts)
+      elif message.kind == MessageKind.ANNOUNCE:
+        self._send_parts(cipher, splitted_pieces)
+    finally:
+      self._send_lock.release()
+
+  def _process(self, message):
+    try:
+      if message.kind == MessageKind.PING:
+        self.send(
+          Message(
+            MessageKind.PONG,
+            uid=message.uid
+          )
+        )
+      elif message.kind == MessageKind.BYE:
+        self.close()
+      elif message.kind == MessageKind.QUERY:
+        if self._queries_pending >= 3 or message.ttl < 1 or self.is_cached(message):
+          self.send(
+            Message(
+              MessageKind.NOTAVAIL,
+              uid=message.uid
+            )
+          )
+
+          return
+
+        self.cache(message)
+
+        piece = self._peer.query(message.hash, message.uid, message.ttl - 1, self)
+        if piece:
+          self.send(
+            Message(
+              MessageKind.QUERY_HIT,
+              uid=message.uid,
+              piece=piece
+            )
+          )
+
+          return
+
+        self.send(
+          Message(
+            MessageKind.NOTAVAIL,
+            uid=message.uid
+          )
+        )
+      elif message.kind == MessageKind.CRAWL:
+        if message.ttl <= 1:
+          self.send(
+            Message(
+              MessageKind.NOTAVAIL,
+              uid=message.uid
+            )
+          )
+
+          return
+
+        if self.is_cached(message):          
+          self.send(
+            Message(
+              MessageKind.MAP,
+              uid=message.uid,
+              map=self._peer.crawl(None, flat=True)
+            )
+          )
+
+          return
+
+        self.cache(message)
+
+        map = self._peer.crawl(message.uid, message.ttl - 1, self)
+        
+        self.send(
+          Message(
+            MessageKind.MAP,
+            uid=message.uid,
+            map=map
+          )
+        )
+      elif message.kind == MessageKind.ANNOUNCE:
+        self.near_pieces.union(message.pieces)    
+    except:
+      self.close()
+
+  def listen(self):
+    try:
+      self.handshake()
+    except:
+      if not self.is_inbound:
+        log.error(f'`{self.addr}` ({self.encoded_uid}): Handshake problem.')
+
+      self.close()
+
+      return
+
+    self._peer.connections.append(self)
+
+    self.announce()
+
+    while self.is_alive:
+      try:
+        message = self.receive()
+      except:
+        self.close()
+
+        return
+
+      if message.kind in (
+        MessageKind.PING,
+        MessageKind.ANNOUNCE
+      ):
+        if self.is_cached(message):
+          continue
+
+        self.cache(message)      
+      elif message.kind in (
+        MessageKind.PONG,
+        MessageKind.QUERY_HIT,
+        MessageKind.NOTAVAIL,
+        MessageKind.MAP
+      ):
+        self._queue.append(message)
+
+        continue
+
+      spawn_thread(self._process, message)
+
+class Interface:
+  def __init__(self, conn, peer=None):
+    self._conn = conn
+    self._peer = peer
+
+  def close(self):
+    self._conn.close()
+
+  def _contact(self, command, data=b''):
+    self._conn.send(command + data)
+
+    response = self._conn.recv()
+
+    return (response[:4], response[4:])
+
+  def save(self, piece):
+    response, data = self._contact(b'SAVE', piece)
+
+    if response == b'SAVD':
+      return data
+
+  def query(self, hash):
+    response, data = self._contact(b'FIND', hash)
+
+    if response == b'QHIT':
+      return data
+
+  def crawl(self):
+    response, data = self._contact(b'CRWL')
+
+    if response == b'NMAP':
+      return Map.create(data)
+
+  def announce(self):
+    response, _ = self._contact(b'ANON')
+
+    return response == b'DONE'
+
+  def listen(self):
+    while True:
+      query = self._conn.recv()
+
+      if len(query) < 4:
+        continue
+
+      command = query[:4]
+      data = query[4:]
+
+      try:
+        if command == b'SAVE':
+          if len(data) < 1 or len(data) > Piece.PIECE_SIZE:
+            raise YAFNError
+
+          key, cipher = salsa20_create_encryptor()
+          checksum = adler32(data)
+
+          data = zlib.compress(data, 9)[2:-4]
+          data = cipher.encrypt(data)
+          piece = Storage.save_piece(data)
+
+          self._conn.send(b'SAVD' + piece.hash + key + struct.pack('!L', checksum))
+        elif command == b'FIND':
+          if len(data) != 32 + 40 + 4:
+            raise YAFNError
+
+          hash = data[:32]
+          key = data[32:72]
+          checksum = struct.unpack('!L', data[72:76])[0]
+
+          piece = self._peer.query(hash, uuid.uuid1().bytes)
+
+          if piece:
+            data = piece.join()
+
+            cipher = salsa20_create_decryptor(key)
+            data = cipher.decrypt(data)
+            data = zlib.decompress(data, -15)
+
+            if adler32(data) != checksum:
+              raise YAFNError
+
+            self._conn.send(b'QHIT' + data)
+
+            continue
+
+          self._conn.send(b'NOTA')
+        elif command == b'CRWL':
+          if len(data) != 0:
+            raise YAFNError
+
+          map = self._peer.crawl(uuid.uuid1().bytes)
+
+          self._conn.send(b'NMAP' + map.dump())
+        elif command == b'ANON':
+          if len(data) != 0:
+            raise YAFNError
+
+          self._peer.announce()
+
+          self._conn.send(b'DONE')
+        else:
+          raise YAFNError
+      except:
+        self._conn.send(b'FAIL')
+
+  @staticmethod
+  def create(peer):
+    context = zmq.Context()
+    conn = context.socket(zmq.REP)
+    
+    while True:
+      try:
+        conn.bind('tcp://*:49872')
+      except:
+        log.error('Failed to bind the interface.')
+
+        time.sleep(10)
+
+        continue
+
+      break
+
+    return Interface(conn, peer)
+
+  @staticmethod
+  def connect():
+    context = zmq.Context()
+    conn = context.socket(zmq.REQ)
+    conn.connect('tcp://127.0.0.1:49872')
+
+    return Interface(conn)
+
+class Tracker:
+  def __init__(self, host):
+    self.host = host
+    self.disabled_for = 0
+    self.disabled = False
+
+  def _request(self, remote_addr=None):
+    try:
+      request = urllib.request.Request(f'http://{self.host}:49873/track')
+      if remote_addr:
+        request.add_header('YAFN-Remote-Address', remote_addr)
+
+      with urllib.request.urlopen(request, timeout=30) as resp:
+        code = resp.getcode()
+        data = resp.read()
+
+        return (code, data)
+    except:
+      return None
+
+    return None
+
+  def _contact(self, remote_addr=None):
+    resp = self._request(remote_addr)
+
+    if not resp or resp[0] != 200:
+      return None
+
+    try:
+      data = cbor2.loads(resp[1])
+
+      if 'remote_addr' not in data\
+      or type(data['remote_addr']) is not str:
+        return None
+
+      if 'is_accessible' not in data\
+      or type(data['is_accessible']) is not bool:
+        return None
+
+      if 'peers' not in data\
+      or type(data['peers']) is not list\
+      or not all(map(
+        lambda peer: type(peer) is dict
+                 and 'address' in peer and type(peer['address']) is str
+                 and 'uid' in peer and type(peer['uid']) is bytes and len(peer['uid']) == 20
+                 and 'latency' in peer and type(peer['latency']) is int and peer['latency'] >= 0
+                 and 'last_check' in peer and type(peer['last_check']) is int and peer['last_check'] >= 0,
+        data['peers']
+      )):
+        return None
+
+      return data
+    except:
+      return None
+
+  def contact(self, remote_addr=None):
+    if self.disabled:
+      if self.disabled_for > 0:
+        self.disabled_for -= 1
+
+        return None
+      else:
+        self.disabled = False
+
+    data = self._contact(remote_addr)
+
+    if not data:
+      self.disabled_for += 1
+
+      if self.disabled_for > 3:
+        self.disabled = True
+
+    return data
+
+class Peer:
+  def __init__(self):
+    self.connections = []
+    self.trackers = []
+
+    self.remote_addr = None
+
+    Storage.setup()
+
+    self.keypair = Storage.get_keypair()
+    self.uid = generate_uid(self.keypair)
+
+    self.pubkey = self.keypair.public_key().export_key('DER')
+
+    trackers = Storage.get_trackers()
+
+    for host in trackers:
+      self.trackers.append(Tracker(host))
+
+  def _discover_peers(self):
+    peers = set()
+
+    for tracker in self.trackers:
+      data = tracker.contact(self.remote_addr)
+
+      if data is None:
+        if tracker.disabled_for <= 1:
+          log.error(f'Tracker `{tracker.host}` contact problem.')
+
+        continue
+
+      if data['is_accessible']:
+        if not self.remote_addr:
+          self.remote_addr = data['remote_addr']
+
+          log.info(f'Remote address: `{self.remote_addr}`.')
+
+      peers = peers.union({
+        peer['address'] for peer in data['peers']
+      })
+
+    return peers
+
+  def _serve(self):
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+    while True:
+      try:
+        s.bind(('0.0.0.0', 49871))
+        s.listen()
+      except:
+        log.error('Failed to bind the port.')
+
+        time.sleep(15)
+
+        continue
+
+      break
+      
+    log.info('Ready to accept incoming connections.')
+
+    while True:
+      try:
+        conn, remote = s.accept()
+      except:
+        continue
+
+      conn = Connection(
+        self,
+        conn,
+        remote[0]
+      )
+      spawn_thread(conn.listen)
+
+  def _connect_to(self, host, reconnect_attempts=0):
+    log.info(f'`{host}`: Connecting...')
+
+    try:
+      s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+      s.connect((host, 49871))
+    except:
+      log.error(f'`{host}`: Connection problem.')
+
+      if reconnect_attempts < 5:
+        time.sleep(10 * max(reconnect_attempts, 1))
+
+        self.connect_to(host, reconnect_attempts + 1)
+
+      return
+
+    conn = Connection(
+      self,
+      s,
+      host,
+      False,
+      reconnect_attempts
+    )
+    conn.listen()
+
+  def _connect_to_everyone(self):
+    if not self.trackers:
+      return
+
+    addrs = [
+      conn.addr for conn in self.connections.copy()
+    ]
+    peers = self._discover_peers()
+    peers = [
+      addr for addr in peers if addr not in addrs
+    ]
+    if not peers:
+      return
+
+    log.info(f'Discovered {len(peers)} peer{"s" if len(peers) != 1 else ""}.')
+
+    for addr in peers:
+      self.connect_to(addr)
+
+  def query(self, hash, mid, ttl=7, came_from=None):
+    piece = Storage.find_piece(hash)
+
+    if piece:
+      return piece
+
+    if ttl < 1:
+      return None
+
+    connections = self.connections.copy()
+    inbound_connections = []
+    outbound_connections = []
+
+    for conn in connections:
+      if came_from and came_from.uid == conn.uid:
+        continue
+
+      if not conn.is_ok:
+        continue
+   
+      if conn.is_inbound:
+        inbound_connections.append(conn)
+      else:
+        outbound_connections.append(conn)
+
+    random.shuffle(inbound_connections)
+    random.shuffle(outbound_connections)
+
+    if came_from and came_from.is_inbound:
+      connections = inbound_connections + outbound_connections
+    else:
+      connections = outbound_connections + inbound_connections
+
+    for conn in connections:
+      if not conn.is_ok:
+        continue
+
+      if hash in conn.near_pieces:
+        piece = conn.query(hash, mid, ttl)
+        if piece:
+          return Storage.save_piece(piece)
+
+    for conn in connections:
+      if not conn.is_ok:
+        continue
+
+      piece = conn.query(hash, mid, ttl)
+      if piece:
+        return Storage.save_piece(piece)
+
+  def crawl(self, mid, ttl=7, came_from=None, flat=False):
+    connections = self.connections.copy()
+    submaps = []
+
+    for conn in connections:
+      if came_from and came_from.uid == conn.uid:
+        continue
+
+      if not conn.is_ok:
+        continue
+
+      if flat:
+        submaps.append(Map(conn.uid, []))
+      else:
+        map = conn.crawl(mid, ttl)
+        if map:
+          submaps.append(map)
+
+    return Map(self.uid, submaps)
+
+  def announce(self):
+    connections = self.connections.copy()
+
+    for conn in connections:
+      if not conn.is_ok:
+        continue
+
+      conn.announce()
+
+  def connect_to(self, addr, reconnect_attempts=0):
+    if addr == self.remote_addr:
+      return
+
+    for conn in self.connections.copy():
+      if conn.addr == addr:
+        return
+
+    spawn_thread(self._connect_to, addr, reconnect_attempts)
+
+  def start(self, remote_addr=None):
+    log.info('Starting up.')
+    log.info(f'Peer UID: {encode_uid(self.uid)}.')
+
+    if remote_addr:
+      self.remote_addr = remote_addr
+
+      log.info(f'Remote address: `{self.remote_addr}`.')
+
+    spawn_thread(self._serve)
+
+    Timer(
+      60*5,
+      self._connect_to_everyone,
+      2
+    ).start()
+
+    Interface.create(self).listen()