ZeroNet/src/Tor/TorManager.py

import logging
import re
import socket
import binascii
import sys
import os
import time
import random
import subprocess
import atexit

import gevent

from Config import config
from Crypt import CryptRsa
from Site import SiteManager
from lib.PySocks import socks
try:
    from gevent.coros import RLock
except:
    from gevent.lock import RLock
from util import helper
from Debug import Debug
from Plugin import PluginManager


@PluginManager.acceptPlugins
class TorManager(object):
    def __init__(self, fileserver_ip=None, fileserver_port=None):
        self.privatekeys = {}  # Onion: Privatekey
        self.site_onions = {}  # Site address: Onion
        self.tor_exe = "tools/tor/tor.exe"
        self.has_meek_bridges = os.path.isfile("tools/tor/PluggableTransports/meek-client.exe")
        self.tor_process = None
        self.log = logging.getLogger("TorManager")
        self.start_onions = None
        self.conn = None
        self.lock = RLock()
        self.starting = True
        self.event_started = gevent.event.AsyncResult()

        if config.tor == "disable":
            self.enabled = False
            self.start_onions = False
            self.setStatus("Disabled")
        else:
            self.enabled = True
            self.setStatus("Waiting")

        if fileserver_port:
            self.fileserver_port = fileserver_port
        else:
            self.fileserver_port = config.fileserver_port

        self.ip, self.port = config.tor_controller.rsplit(":", 1)
        self.port = int(self.port)

        self.proxy_ip, self.proxy_port = config.tor_proxy.rsplit(":", 1)
        self.proxy_port = int(self.proxy_port)

    def start(self):
        self.log.debug("Starting (Tor: %s)" % config.tor)
        self.starting = True
        try:
            if not self.connect():
                raise Exception("No connection")
            self.log.debug("Tor proxy port %s check ok" % config.tor_proxy)
        except Exception, err:
            if sys.platform.startswith("win"):
                self.log.info(u"Starting self-bundled Tor, due to Tor proxy port %s check error: %s" % (config.tor_proxy, err))
            else:
                self.log.info(u"Disabling Tor, because error while accessing Tor proxy at port %s: %s" % (config.tor_proxy, err))
            self.enabled = False
            # Change to self-bundled Tor ports
            from lib.PySocks import socks
            self.port = 49051
            self.proxy_port = 49050
            socks.setdefaultproxy(socks.PROXY_TYPE_SOCKS5, "127.0.0.1", self.proxy_port)
            if os.path.isfile(self.tor_exe):  # Already, downloaded: sync mode
                self.startTor()
            else:  # Not downloaded yet: Async mode
                gevent.spawn(self.startTor)

    def setStatus(self, status):
        self.status = status
        if "ui_server" in dir(sys.modules.get("main", {})):
            sys.modules["main"].ui_server.updateWebsocket()

    def startTor(self):
        if sys.platform.startswith("win"):
            try:
                if not os.path.isfile(self.tor_exe):
                    self.downloadTor()

                self.log.info("Starting Tor client %s..." % self.tor_exe)
                tor_dir = os.path.dirname(self.tor_exe)
                startupinfo = subprocess.STARTUPINFO()
                startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW
                cmd = r"%s -f torrc --defaults-torrc torrc-defaults --ignore-missing-torrc" % self.tor_exe
                if config.tor_use_bridges:
                    cmd += " --UseBridges 1"

                self.tor_process = subprocess.Popen(cmd, cwd=tor_dir, close_fds=True, startupinfo=startupinfo)
                for wait in range(1, 10):  # Wait for startup
                    time.sleep(wait * 0.5)
                    self.enabled = True
                    if self.connect():
                        if self.isSubprocessRunning():
                            self.request("TAKEOWNERSHIP")  # Shut down Tor client when controll connection closed
                        break
                # Terminate on exit
                atexit.register(self.stopTor)
            except Exception, err:
                self.log.error(u"Error starting Tor client: %s" % Debug.formatException(str(err).decode("utf8", "ignore")))
                self.enabled = False
        self.starting = False
        self.event_started.set(False)
        return False

    def isSubprocessRunning(self):
        return self.tor_process and self.tor_process.pid and self.tor_process.poll() is None

    def stopTor(self):
        self.log.debug("Stopping...")
        try:
            if self.isSubprocessRunning():
                self.request("SIGNAL SHUTDOWN")
        except Exception, err:
            self.log.error("Error stopping Tor: %s" % err)

    def downloadTor(self):
        self.log.info("Downloading Tor...")
        # Check Tor webpage for link
        download_page = helper.httpRequest("https://www.torproject.org/download/download.html").read()
        download_url = re.search('href="(.*?tor.*?win32.*?zip)"', download_page).group(1)
        if not download_url.startswith("http"):
            download_url = "https://www.torproject.org/download/" + download_url

        # Download Tor client
        self.log.info("Downloading %s" % download_url)
        data = helper.httpRequest(download_url, as_file=True)
        data_size = data.tell()

        # Handle redirect
        if data_size < 1024 and "The document has moved" in data.getvalue():
            download_url = re.search('href="(.*?tor.*?win32.*?zip)"', data.getvalue()).group(1)
            data = helper.httpRequest(download_url, as_file=True)
            data_size = data.tell()

        if data_size > 1024:
            import zipfile
            zip = zipfile.ZipFile(data)
            self.log.info("Unpacking Tor")
            for inner_path in zip.namelist():
                if ".." in inner_path:
                    continue
                dest_path = inner_path
                dest_path = re.sub("^Data/Tor/", "tools/tor/data/", dest_path)
                dest_path = re.sub("^Data/", "tools/tor/data/", dest_path)
                dest_path = re.sub("^Tor/", "tools/tor/", dest_path)
                dest_dir = os.path.dirname(dest_path)
                if dest_dir and not os.path.isdir(dest_dir):
                    os.makedirs(dest_dir)

                if dest_dir != dest_path.strip("/"):
                    data = zip.read(inner_path)
                    if not os.path.isfile(dest_path):
                        open(dest_path, 'wb').write(data)
        else:
            self.log.error("Bad response from server: %s" % data.getvalue())
            return False

    def connect(self):
        if not self.enabled:
            return False
        self.site_onions = {}
        self.privatekeys = {}

        return self.connectController()

    def connectController(self):
        if "socket_noproxy" in dir(socket):  # Socket proxy-patched, use non-proxy one
            conn = socket.socket_noproxy(socket.AF_INET, socket.SOCK_STREAM)
        else:
            conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

        self.log.info("Connecting to Tor Controller %s:%s" % (self.ip, self.port))
        try:
            with self.lock:
                conn.connect((self.ip, self.port))

                # Auth cookie file
                res_protocol = self.send("PROTOCOLINFO", conn)
                cookie_match = re.search('COOKIEFILE="(.*?)"', res_protocol)

                if config.tor_password:
                    res_auth = self.send('AUTHENTICATE "%s"' % config.tor_password, conn)
                elif cookie_match:
                    cookie_file = cookie_match.group(1).decode("string-escape")
                    auth_hex = binascii.b2a_hex(open(cookie_file, "rb").read())
                    res_auth = self.send("AUTHENTICATE %s" % auth_hex, conn)
                else:
                    res_auth = self.send("AUTHENTICATE", conn)

                assert "250 OK" in res_auth, "Authenticate error %s" % res_auth

                # Version 0.2.7.5 required because ADD_ONION support
                res_version = self.send("GETINFO version", conn)
                version = re.search('version=([0-9\.]+)', res_version).group(1)
                assert float(version.replace(".", "0", 2)) >= 207.5, "Tor version >=0.2.7.5 required, found: %s" % version

                self.setStatus(u"Connected (%s)" % res_auth)
                self.event_started.set(True)
                self.starting = False
                self.connecting = False
                self.conn = conn
        except Exception, err:
            self.conn = None
            self.setStatus(u"Error (%s)" % str(err).decode("utf8", "ignore"))
            self.log.error(u"Tor controller connect error: %s" % Debug.formatException(str(err).decode("utf8", "ignore")))
            self.enabled = False
        return self.conn

    def disconnect(self):
        self.conn.close()
        self.conn = None

    def startOnions(self):
        if self.enabled:
            self.log.debug("Start onions")
            self.start_onions = True
            self.getOnion("global")

    # Get new exit node ip
    def resetCircuits(self):
        res = self.request("SIGNAL NEWNYM")
        if "250 OK" not in res:
            self.setStatus(u"Reset circuits error (%s)" % res)
            self.log.error("Tor reset circuits error: %s" % res)

    def addOnion(self):
        if len(self.privatekeys) >= config.tor_hs_limit:
            return random.choice([key for key in self.privatekeys.keys() if key != self.site_onions.get("global")])

        result = self.makeOnionAndKey()
        if result:
            onion_address, onion_privatekey = result
            self.privatekeys[onion_address] = onion_privatekey
            self.setStatus(u"OK (%s onions running)" % len(self.privatekeys))
            SiteManager.peer_blacklist.append((onion_address + ".onion", self.fileserver_port))
            return onion_address
        else:
            return False

    def makeOnionAndKey(self):
        res = self.request("ADD_ONION NEW:RSA1024 port=%s" % self.fileserver_port)
        match = re.search("ServiceID=([A-Za-z0-9]+).*PrivateKey=RSA1024:(.*?)[\r\n]", res, re.DOTALL)
        if match:
            onion_address, onion_privatekey = match.groups()
            return (onion_address, onion_privatekey)
        else:
            self.setStatus(u"AddOnion error (%s)" % res)
            self.log.error("Tor addOnion error: %s" % res)
            return False

    def delOnion(self, address):
        res = self.request("DEL_ONION %s" % address)
        if "250 OK" in res:
            del self.privatekeys[address]
            self.setStatus("OK (%s onion running)" % len(self.privatekeys))
            return True
        else:
            self.setStatus(u"DelOnion error (%s)" % res)
            self.log.error("Tor delOnion error: %s" % res)
            self.disconnect()
            return False

    def request(self, cmd):
        with self.lock:
            if not self.enabled:
                return False
            if not self.conn:
                if not self.connect():
                    return ""
            return self.send(cmd)

    def send(self, cmd, conn=None):
        if not conn:
            conn = self.conn
        self.log.debug("> %s" % cmd)
        back = ""
        for retry in range(2):
            try:
                conn.sendall("%s\r\n" % cmd)
                while not back.endswith("250 OK\r\n"):
                    back += conn.recv(1024 * 64).decode("utf8", "ignore")
                break
            except Exception, err:
                self.log.error("Tor send error: %s, reconnecting..." % err)
                self.disconnect()
                time.sleep(1)
                self.connect()
                back = None
        if back:
            self.log.debug("< %s" % back.strip())
        return back

    def getPrivatekey(self, address):
        return self.privatekeys[address]

    def getPublickey(self, address):
        return CryptRsa.privatekeyToPublickey(self.privatekeys[address])

    def getOnion(self, site_address):
        if not self.enabled:
            return None

        if config.tor == "always":  # Different onion for every site
            onion = self.site_onions.get(site_address)
        else:  # Same onion for every site
            onion = self.site_onions.get("global")
            site_address = "global"

        if not onion:
            with self.lock:
                self.site_onions[site_address] = self.addOnion()
                onion = self.site_onions[site_address]
                self.log.debug("Created new hidden service for %s: %s" % (site_address, onion))

        return onion

    # Creates and returns a
    # socket that has connected to the Tor Network
    def createSocket(self, onion, port):
        if not self.enabled:
            return False
        self.log.debug("Creating new Tor socket to %s:%s" % (onion, port))
        if self.starting:
            self.log.debug("Waiting for startup...")
            self.event_started.get()
        if config.tor == "always":  # Every socket is proxied by default, in this mode
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        else:
            sock = socks.socksocket()
            sock.set_proxy(socks.SOCKS5, self.proxy_ip, self.proxy_port)
        return sock