def process_from_address(addr, port, torstate=None):
    """
    Determines the PID from the address/port provided by using lsof
    and returns it as an int (or None if it couldn't be
    determined). In the special case the addr is '(Tor_internal)' then
    the PID of the Tor process (as gotten from the torstate object) is
    returned (or 0 if unavailable, e.g. a Tor which doesn't implement
    'GETINFO process/pid'). In this case if no TorState instance is
    given, None is returned.
    """

    if addr is None:
        return None

    if "(tor_internal)" == str(addr).lower():
        if torstate is None:
            return None
        return int(torstate.tor_pid)

    proc = subprocess.Popen(['lsof', '-i', '4tcp@%s:%s' % (addr, port)],
                            stdout=subprocess.PIPE)
    (stdout, stderr) = proc.communicate()
    lines = stdout.split(b'\n')
    if len(lines) > 1:
        return int(lines[1].split()[1]) def delete_file_or_tree(*args):
    """
    For every path in args, try to delete it as a file or a directory
    tree. Ignores deletion errors.
    """

    for f in args:
        try:
            os.unlink(f)
        except OSError:
            shutil.rmtree(f, ignore_errors=True) def find_keywords(args, key_filter=lambda x: not x.startswith("$")):
    """
    This splits up strings like name=value, foo=bar into a dict. Does NOT deal
    with quotes in value (e.g. key="value with space" will not work

    By default, note that it takes OUT any key which starts with $ (i.e. a
    single dollar sign) since for many use-cases the way Tor encodes nodes
    with "$hash=name" looks like a keyword argument (but it isn't). If you
    don't want this, override the "key_filter" argument to this method.

    :param args: a list of strings, each with one key=value pair

    :return:
        a dict of key->value (both strings) of all name=value type
        keywords found in args.
    """
    filtered = [x for x in args if '=' in x and key_filter(x.split('=')[0])]
    return dict(x.split('=', 1) for x in filtered) def find_tor_binary(globs=('/usr/sbin/', '/usr/bin/',
                           '/Applications/TorBrowser_*.app/Contents/MacOS/'),
                    system_tor=True):
    """
    Tries to find the tor executable using the shell first or in in the
    paths whose glob-patterns is in the given 'globs'-tuple.

    :param globs:
        A tuple of shell-style globs of directories to use to find tor
        (TODO consider making that globs to actual tor binary?)

    :param system_tor:
        This controls whether bash is used to seach for 'tor' or
        not. If False, we skip that check and use only the 'globs'
        tuple.
    """

    # Try to find the tor executable using the shell
    if system_tor:
        try:
            proc = subprocess.Popen(
                ('which tor'),
                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                shell=True
            )
        except OSError:
            pass
        else:
            stdout, _ = proc.communicate()
            if proc.poll() == 0 and stdout != '':
                return stdout.strip()

    # the shell may not provide type and tor is usually not on PATH when using
    # the browser-bundle. Look in specific places
    for pattern in globs:
        for path in glob.glob(pattern):
            torbin = os.path.join(path, 'tor')
            if is_executable(torbin):
                return torbin
    return None def maybe_ip_addr(addr):
    """
    Tries to return an IPAddress, otherwise returns a string.

    TODO consider explicitly checking for .exit or .onion at the end?
    """

    if six.PY2 and isinstance(addr, str):
        addr = unicode(addr)  # noqa
    try:
        return ipaddress.ip_address(addr)
    except ValueError:
        pass
    return str(addr) def unescape_quoted_string(string):
    r'''
    This function implementes the recommended functionality described in the
    tor control-spec to be compatible with older tor versions:

      * Read \\n \\t \\r and \\0 ... \\377 as C escapes.
      * Treat a backslash followed by any other character as that character.

    Except the legacy support for the escape sequences above this function
    implements parsing of QuotedString using qcontent from

    QuotedString = DQUOTE *qcontent DQUOTE

    :param string: The escaped quoted string.
    :returns: The unescaped string.
    :raises ValueError: If the string is in a invalid form
                        (e.g. a single backslash)
    '''
    match = re.match(r'''^"((?:[^"\\]|\\.)*)"$''', string)
    if not match:
        raise ValueError("Invalid quoted string", string)
    string = match.group(1)
    # remove backslash before all characters which should not be
    # handeled as escape codes by string.decode('string-escape').
    # This is needed so e.g. '\x00' is not unescaped as '\0'
    string = re.sub(r'((?:^|[^\\])(?:\\\\)*)\\([^ntr0-7\\])', r'\1\2', string)
    if six.PY3:
        # XXX hmmm?
        return bytes(string, 'ascii').decode('unicode-escape')
    return string.decode('string-escape') @defer.inlineCallbacks
def available_tcp_port(reactor):
    """
    Returns a Deferred firing an available TCP port on localhost.
    It does so by listening on port 0; then stopListening and fires the
    assigned port number.
    """

    endpoint = serverFromString(reactor, 'tcp:0:interface=127.0.0.1')
    port = yield endpoint.listen(NoOpProtocolFactory())
    address = port.getHost()
    yield port.stopListening()
    defer.returnValue(address.port) def version_at_least(version_string, major, minor, micro, patch):
    """
    This returns True if the version_string represents a Tor version
    of at least ``major``.``minor``.``micro``.``patch`` version,
    ignoring any trailing specifiers.
    """
    parts = re.match(
        r'^([0-9]+)\.([0-9]+)\.([0-9]+)\.([0-9]+).*$',
        version_string,
    )
    for ver, gold in zip(parts.group(1, 2, 3, 4), (major, minor, micro, patch)):
        if int(ver) < int(gold):
            return False
        elif int(ver) > int(gold):
            return True
    return True def default_control_port():
    """
    This returns a default control port, which respects an environment
    variable `TX_CONTROL_PORT`. Without the environment variable, this
    returns 9151 (the Tor Browser Bundle default).

    You shouldn't use this in "normal" code, this is a convenience for
    the examples.
    """
    try:
        return int(os.environ['TX_CONTROL_PORT'])
    except KeyError:
        return 9151 class _Listener(object):
    """
    Internal helper.
    """

    def __init__(self):
        self._listeners = set()

    def add(self, callback):
        """
        Add a callback to this listener
        """
        self._listeners.add(callback)

    __call__ = add  #: alias for "add"

    def remove(self, callback):
        """
        Remove a callback from this listener
        """
        self._listeners.remove(callback)

    def notify(self, *args, **kw):
        """
        Calls all listeners with the specified args.

        Returns a Deferred which callbacks when all the listeners
        which return Deferreds have themselves completed.
        """
        calls = []

        def failed(fail):
            # XXX use logger
            fail.printTraceback()

        for cb in self._listeners:
            d = defer.maybeDeferred(cb, *args, **kw)
            d.addCallback(maybe_coroutine)
            d.addErrback(failed)
            calls.append(d)
        return defer.DeferredList(calls) class _ListenerCollection(object):
    """
    Internal helper.

    This collects all your valid event listeners together in one
    object if you want.
    """
    def __init__(self, valid_events):
        self._valid_events = valid_events
        for e in valid_events:
            setattr(self, e, _Listener())

    def __call__(self, event, callback):
        if event not in self._valid_events:
            raise Exception("Invalid event '{}'".format(event))
        getattr(self, event).add(callback)

    def remove(self, event, callback):
        if event not in self._valid_events:
            raise Exception("Invalid event '{}'".format(event))
        getattr(self, event).remove(callback)

    def notify(self, event, *args, **kw):
        if event not in self._valid_events:
            raise Exception("Invalid event '{}'".format(event))
        getattr(self, event).notify(*args, **kw) def create_tbb_web_headers():
    """
    Returns a new `twisted.web.http_headers.Headers` instance
    populated with tags to mimic Tor Browser. These include values for
    `User-Agent`, `Accept`, `Accept-Language` and `Accept-Encoding`.
    """
    return Headers({
        b"User-Agent": [b"Mozilla/5.0 (Windows NT 6.1; rv:45.0) Gecko/20100101 Firefox/45.0"],
        b"Accept": [b"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"],
        b"Accept-Language": [b"en-US,en;q=0.5"],
        b"Accept-Encoding": [b"gzip, deflate"],
    }) class FakeControlProtocol(object):
    """
    This is a little weird, but in most tests the answer at the top of
    the list is sent back immediately in an already-called
    Deferred. However, if the answer list is empty at the time of the
    call, instead the returned Deferred is added to the pending list
    and answer_pending() may be called to have the next Deferred
    fire. (see test_accept_all_postbootstrap for an example).

    It is done this way in case we need to have some other code run
    between the get_conf (or whatever) and the callback -- if the
    Deferred is already-fired when get_conf runs, there's a Very Good
    Chance (always?) that the callback just runs right away.
    """

    def __init__(self, answers):
        self.answers = answers
        self.pending = []
        self.post_bootstrap = defer.succeed(self)
        self.on_disconnect = defer.Deferred()
        self.sets = []
        self.events = {}  #: event type -> callback
        self.pending_events = {}  #: event type -> list
        self.is_owned = -1
        self.version = "0.2.8.0"
        # XXX can we get rud of 'commands' and just use pending?
        self.commands = []

    def queue_command(self, cmd):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            self.commands.append((cmd, d))
            return d

        d = defer.succeed(self.answers[0])
        self.commands.append((cmd, d))
        self.answers = self.answers[1:]
        return d

    def event_happened(self, event_type, *args):
        '''
        Use this in your tests to send 650 events when an event-listener
        is added.  XXX Also if we've *already* added one? Do that if
        there's a use-case for it
        '''
        if event_type in self.events:
            self.events[event_type](*args)
        elif event_type in self.pending_events:
            self.pending_events[event_type].append(args)
        else:
            self.pending_events[event_type] = [args]

    def answer_pending(self, answer):
        d = self.pending[0]
        self.pending = self.pending[1:]
        d.callback(answer)

    def get_info(self, info):
        a = self.answers.pop()
        return defer.succeed(a)

    def get_info_raw(self, info):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            return d

        d = defer.succeed(self.answers[0])
        self.answers = self.answers[1:]
        return d

    @defer.inlineCallbacks
    def get_info_incremental(self, info, cb):
        text = yield self.get_info_raw(info)
        for line in text.split('\r\n'):
            cb(line)
        defer.returnValue('')  # FIXME uh....what's up at torstate.py:350?

    def get_conf(self, info):
        if len(self.answers) == 0:
            d = defer.Deferred()
            self.pending.append(d)
            return d

        d = defer.succeed(self.answers[0])
        self.answers = self.answers[1:]
        return d

    get_conf_raw = get_conf  # up to test author ensure the answer is a raw string
    get_conf_single = get_conf

    def set_conf(self, *args):
        for i in range(0, len(args), 2):
            self.sets.append((args[i], args[i + 1]))
        return defer.succeed('')

    def add_event_listener(self, nm, cb):
        self.events[nm] = cb
        if nm in self.pending_events:
            for event in self.pending_events[nm]:
                cb(*event)
        return defer.succeed(None)

    def remove_event_listener(self, nm, cb):
        del self.events[nm]
import os
import sys
import tempfile
import ipaddress
from mock import patch
from unittest import skipIf
from os.path import exists
from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet.endpoints import TCP4ServerEndpoint
from twisted.internet.interfaces import IProtocolFactory
from zope.interface import implementer
from txtorcon.util import process_from_address
from txtorcon.util import delete_file_or_tree
from txtorcon.util import find_keywords
from txtorcon.util import find_tor_binary
from txtorcon.util import maybe_ip_addr
from txtorcon.util import unescape_quoted_string
from txtorcon.util import available_tcp_port
from txtorcon.util import version_at_least
from txtorcon.util import default_control_port
from txtorcon.util import _Listener, _ListenerCollection
from txtorcon.util import create_tbb_web_headers
from txtorcon.testutil import FakeControlProtocol
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from txtorcon import util
        from twisted.internet import reactor




class FakeState:
    tor_pid = 0


@implementer(IProtocolFactory)
class FakeProtocolFactory:

    def doStart(self):
        "IProtocolFactory API"

    def doStop(self):
        "IProtocolFactory API"

    def buildProtocol(self, addr):
        "IProtocolFactory API"
        return None


class TestGeoIpDatabaseLoading(unittest.TestCase):

    def test_bad_geoip_path(self):
        "fail gracefully if a db is missing"
        self.assertRaises(IOError, util.create_geoip, '_missing_path_')

    def test_missing_geoip_module(self):
        "return none if geoip module is missing"
        _GeoIP = util.GeoIP
        util.GeoIP = None
        (fd, f) = tempfile.mkstemp()
        ret_val = util.create_geoip(f)
        delete_file_or_tree(f)
        util.GeoIP = _GeoIP
        self.assertEqual(ret_val, None)

    @skipIf('pypy' in sys.version.lower(), "No GeoIP in PyPy")
    def test_return_geoip_object(self):
        # requires a valid GeoIP database to work, so hopefully we're
        # on Debian or similar...
        fname = "/usr/share/GeoIP/GeoIP.dat"
        if not exists(fname):
            return

        ret_val = util.create_geoip(fname)
        self.assertEqual(type(ret_val).__name__, 'GeoIP')


class TestFindKeywords(unittest.TestCase):

    def test_filter(self):
        "make sure we filter out keys that look like router IDs"
        self.assertEqual(

            find_keywords("foo=bar $1234567890=routername baz=quux".split()),