1164

Committed 2 Oct 2018 - 22:05 coverage increased (+0.0002%) to 99.957%

Build # 1164

Build Type

Pull #316

travis-ci

Committed by

web-flow

Commit Message

fix asserts

Pull Request Pull Request #316: Ticket313 pivate key file

Run Details

21 of 21 new or added lines in 1 file covered. (100.0%)

8 existing lines in 3 files now uncovered.

4648 of 4650 relevant lines covered (99.96%)

11.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.17

/txtorcon/util.py

# -*- coding: utf-8 -*-

from __future__ import absolute_import
from __future__ import print_function
from __future__ import with_statement

import glob
import os
import hmac
import hashlib
import shutil
import socket
import subprocess
import ipaddress
import struct
import re
import six

from twisted.internet import defer
from twisted.internet.interfaces import IProtocolFactory
from twisted.internet.endpoints import serverFromString
from twisted.web.http_headers import Headers

from zope.interface import implementer
from zope.interface import Interface

if six.PY3:
    import asyncio

try:
    import GeoIP as _GeoIP
    GeoIP = _GeoIP
except ImportError:
    GeoIP = None

city = None
country = None
asn = None


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"],
    })


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 create_geoip(fname):
    # It's more "pythonic" to just wait for the exception,
    # but GeoIP prints out "Can't open..." messages for you,
    # which isn't desired here
    if not os.path.isfile(fname):
        raise IOError("Can't find %s" % fname)

    if GeoIP is None:
        return None

    # just letting any errors make it out
    return GeoIP.open(fname, GeoIP.GEOIP_STANDARD)


def maybe_create_db(path):
    try:
        return create_geoip(path)
    except IOError:
        return None


city = maybe_create_db("/usr/share/GeoIP/GeoLiteCity.dat")
asn = maybe_create_db("/usr/share/GeoIP/GeoIPASNum.dat")
country = maybe_create_db("/usr/share/GeoIP/GeoIP.dat")


def is_executable(path):
    """Checks if the given path points to an existing, executable file"""
    return os.path.isfile(path) and os.access(path, os.X_OK)


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 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 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 ip_from_int(ip):
        """ Convert long int back to dotted quad string """
        return socket.inet_ntoa(struct.pack('>I', ip))


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 hmac_sha256(key, msg):
    """
    Adapted from rransom's tor-utils git repository. Returns the
    digest (binary) of an HMAC with SHA256 over msg with key.
    """

    return hmac.new(key, msg, hashlib.sha256).digest()


CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE = os.urandom(32)


def compare_via_hash(x, y):
    """
    Taken from rransom's tor-utils git repository, to compare two
    hashes in something resembling constant time (or at least, not
    leaking timing info?)
    """
    return (hmac_sha256(CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE, x) ==
            hmac_sha256(CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE, y))


class NetLocation(object):
    """
    Represents the location of an IP address, either city or country
    level resolution depending on what GeoIP database was loaded. If
    the ASN database is available you get that also.
    """

    def __init__(self, ipaddr):
        "ipaddr should be a dotted-quad"
        self.ip = ipaddr
        self.latlng = (None, None)
        self.countrycode = None
        self.city = None
        self.asn = None

        if self.ip is None or self.ip == 'unknown':
            return

        if city:
            try:
                r = city.record_by_addr(self.ip)
            except Exception:
                r = None
            if r is not None:
                self.countrycode = r['country_code']
                self.latlng = (r['latitude'], r['longitude'])
                try:
                    self.city = (r['city'], r['region_code'])
                except KeyError:
                    self.city = (r['city'], r['region_name'])

        elif country:
            self.countrycode = country.country_code_by_addr(ipaddr)

        else:
            self.countrycode = ''

        if asn:
            try:
                self.asn = asn.org_by_addr(self.ip)
            except Exception:
                self.asn = None


@implementer(IProtocolFactory)
class NoOpProtocolFactory:
    """
    This is an IProtocolFactory that does nothing. Used for testing,
    and for :method:`available_tcp_port`
    """
    def noop(self, *args, **kw):
        pass
    buildProtocol = noop
    doStart = noop
    doStop = noop


@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 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')


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 IListener(Interface):
    def add(callback):
        """
        Add a listener. The arguments to the callback are determined by whomever calls notify()
        """

    def remove(callback):
        """
        Add a listener. The arguments to the callback are determined by whomever calls notify()
        """

    def notify(*args, **kw):
        """
        Calls every listener with the given args and keyword-args.

        XXX errors? just log?
        """


def maybe_coroutine(obj):
    """
    If 'obj' is a coroutine and we're using Python3, wrap it in
    ensureDeferred. Otherwise return the original object.

    (This is to insert in all callback chains from user code, in case
    that user code is Python3 and used 'async def')
    """
    if six.PY3 and asyncio.iscoroutine(obj):
        return defer.ensureDeferred(obj)
    return obj


@implementer(IListener)
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)


# similar to OneShotObserverList in Tahoe-LAFS
class SingleObserver(object):
    """
    A helper for ".when_*()" sort of functions.
    """
    _NotFired = object()

    def __init__(self):
        self._observers = []
        self._fired = self._NotFired

    def when_fired(self):
        d = defer.Deferred()
        if self._fired is not self._NotFired:
            d.callback(self._fired)
        else:
            self._observers.append(d)
        return d

    def fire(self, value):
        if self._observers is None:
            return  # raise RuntimeError("already fired") ?
        self._fired = value
        for d in self._observers:
            d.callback(self._fired)
        self._observers = None
        return value  # so we're transparent if used as a callback


class _Version(object):
    """
    Replacement for incremental.Version until
    https://github.com/meejah/txtorcon/issues/233 and/or
    https://github.com/hawkowl/incremental/issues/31 is fixed.
    """
    # as of latest incremental, it should only access .package and
    # .short() via the getVersionString() method that Twisted's
    # deprecated() uses...

    def __init__(self, package, major, minor, patch):
        self.package = package
        self.major = major
        self.minor = minor
        self.patch = patch

    def short(self):
        return '{}.{}.{}'.format(self.major, self.minor, self.patch)


# originally from magic-wormhole code
def _is_non_public_numeric_address(host):
    """
    returns True if 'host' is not public
    """
    # for numeric hostnames, skip RFC1918 addresses, since no Tor exit
    # node will be able to reach those. Likewise ignore IPv6 addresses.
    try:
        a = ipaddress.ip_address(six.text_type(host))
    except ValueError:
        return False        # non-numeric, let Tor try it
    if a.is_loopback or a.is_multicast or a.is_private or a.is_reserved \
       or a.is_unspecified:
        return True         # too weird, don't connect
    return False

1	# -- coding: utf-8 --
2
3	from __future__ import absolute_import	12×
4	from __future__ import print_function	12×
5	from __future__ import with_statement	12×
6
7	import glob	12×
8	import os	12×
9	import hmac	12×
10	import hashlib	12×
11	import shutil	12×
12	import socket	12×
13	import subprocess	12×
14	import ipaddress	12×
15	import struct	12×
16	import re	12×
17	import six	12×
18
19	from twisted.internet import defer	12×
20	from twisted.internet.interfaces import IProtocolFactory	12×
21	from twisted.internet.endpoints import serverFromString	12×
22	from twisted.web.http_headers import Headers	12×
23
24	from zope.interface import implementer	12×
25	from zope.interface import Interface	12×
26
27	if six.PY3:	12×
UNCOV 28	import asyncio	6×
29
30	try:	12×
31	import GeoIP as _GeoIP	12×
32	GeoIP = _GeoIP	12×
33	except ImportError:	3×
34	GeoIP = None	3×
35
36	city = None	12×
37	country = None	12×
38	asn = None	12×
39
40
41	def create_tbb_web_headers():	12×
42	"""
43	Returns a new `twisted.web.http_headers.Headers` instance
44	populated with tags to mimic Tor Browser. These include values for
45	`User-Agent`, `Accept`, `Accept-Language` and `Accept-Encoding`.
46	"""
47	return Headers({	12×
48	b"User-Agent": [b"Mozilla/5.0 (Windows NT 6.1; rv:45.0) Gecko/20100101 Firefox/45.0"],
49	b"Accept": [b"text/html,application/xhtml+xml,application/xml;q=0.9,/;q=0.8"],
50	b"Accept-Language": [b"en-US,en;q=0.5"],
51	b"Accept-Encoding": [b"gzip, deflate"],
52	})
53
54
55	def version_at_least(version_string, major, minor, micro, patch):	12×
56	"""
57	This returns True if the version_string represents a Tor version
58	of at least ``major``.``minor``.``micro``.``patch`` version,
59	ignoring any trailing specifiers.
60	"""
61	parts = re.match(	12×
62	r'^([0-9]+)\.([0-9]+)\.([0-9]+)\.([0-9]+).*$',
63	version_string,
64	)
65	for ver, gold in zip(parts.group(1, 2, 3, 4), (major, minor, micro, patch)):	12×
66	if int(ver) < int(gold):	12×
67	return False	12×
68	elif int(ver) > int(gold):	12×
69	return True	12×
70	return True	12×
71
72
73	def create_geoip(fname):	12×
74	# It's more "pythonic" to just wait for the exception,
75	# but GeoIP prints out "Can't open..." messages for you,
76	# which isn't desired here
77	if not os.path.isfile(fname):	12×
78	raise IOError("Can't find %s" % fname)	12×
79
80	if GeoIP is None:	12×
81	return None	12×
82
83	# just letting any errors make it out
84	return GeoIP.open(fname, GeoIP.GEOIP_STANDARD)	12×
85
86
87	def maybe_create_db(path):	12×
88	try:	12×
89	return create_geoip(path)	12×
90	except IOError:	12×
91	return None	12×
92
93
94	city = maybe_create_db("/usr/share/GeoIP/GeoLiteCity.dat")	12×
95	asn = maybe_create_db("/usr/share/GeoIP/GeoIPASNum.dat")	12×
96	country = maybe_create_db("/usr/share/GeoIP/GeoIP.dat")	12×
97
98
99	def is_executable(path):	12×
100	"""Checks if the given path points to an existing, executable file"""
101	return os.path.isfile(path) and os.access(path, os.X_OK)	12×
102
103
104	def find_tor_binary(globs=('/usr/sbin/', '/usr/bin/',	12×
105	'/Applications/TorBrowser_*.app/Contents/MacOS/'),
106	system_tor=True):
107	"""
108	Tries to find the tor executable using the shell first or in in the
109	paths whose glob-patterns is in the given 'globs'-tuple.
110
111	:param globs:
112	A tuple of shell-style globs of directories to use to find tor
113	(TODO consider making that globs to actual tor binary?)
114
115	:param system_tor:
116	This controls whether bash is used to seach for 'tor' or
117	not. If False, we skip that check and use only the 'globs'
118	tuple.
119	"""
120
121	# Try to find the tor executable using the shell
122	if system_tor:	12×
123	try:	12×
124	proc = subprocess.Popen(	12×
125	('which tor'),
126	stdout=subprocess.PIPE, stderr=subprocess.PIPE,
127	shell=True
128	)
129	except OSError:	12×
130	pass	12×
131	else:
132	stdout, _ = proc.communicate()	12×
133	if proc.poll() == 0 and stdout != '':	12×
134	return stdout.strip()	!
135
136	# the shell may not provide type and tor is usually not on PATH when using
137	# the browser-bundle. Look in specific places
138	for pattern in globs:	12×
139	for path in glob.glob(pattern):	12×
140	torbin = os.path.join(path, 'tor')	12×
141	if is_executable(torbin):	12×
142	return torbin	!
143	return None	12×
144
145
146	def maybe_ip_addr(addr):	12×
147	"""
148	Tries to return an IPAddress, otherwise returns a string.
149
150	TODO consider explicitly checking for .exit or .onion at the end?
151	"""
152
153	if six.PY2 and isinstance(addr, str):	12×
154	addr = unicode(addr) # noqa	6×
155	try:	12×
156	return ipaddress.ip_address(addr)	12×
157	except ValueError:	12×
158	pass	12×
159	return str(addr)	12×
160
161
162	def find_keywords(args, key_filter=lambda x: not x.startswith("$")):	12×
163	"""
164	This splits up strings like name=value, foo=bar into a dict. Does NOT deal
165	with quotes in value (e.g. key="value with space" will not work
166
167	By default, note that it takes OUT any key which starts with $ (i.e. a
168	single dollar sign) since for many use-cases the way Tor encodes nodes
169	with "$hash=name" looks like a keyword argument (but it isn't). If you
170	don't want this, override the "key_filter" argument to this method.
171
172	:param args: a list of strings, each with one key=value pair
173
174	:return:
175	a dict of key->value (both strings) of all name=value type
176	keywords found in args.
177	"""
178	filtered = [x for x in args if '=' in x and key_filter(x.split('=')[0])]	12×
179	return dict(x.split('=', 1) for x in filtered)	12×
180
181
182	def delete_file_or_tree(*args):	12×
183	"""
184	For every path in args, try to delete it as a file or a directory
185	tree. Ignores deletion errors.
186	"""
187
188	for f in args:	12×
189	try:	12×
190	os.unlink(f)	12×
191	except OSError:	12×
192	shutil.rmtree(f, ignore_errors=True)	12×
193
194
195	def ip_from_int(ip):	12×
196	""" Convert long int back to dotted quad string """
197	return socket.inet_ntoa(struct.pack('>I', ip))	12×
198
199
200	def process_from_address(addr, port, torstate=None):	12×
201	"""
202	Determines the PID from the address/port provided by using lsof
203	and returns it as an int (or None if it couldn't be
204	determined). In the special case the addr is '(Tor_internal)' then
205	the PID of the Tor process (as gotten from the torstate object) is
206	returned (or 0 if unavailable, e.g. a Tor which doesn't implement
207	'GETINFO process/pid'). In this case if no TorState instance is
208	given, None is returned.
209	"""
210
211	if addr is None:	12×
212	return None	12×
213
214	if "(tor_internal)" == str(addr).lower():	12×
215	if torstate is None:	12×
216	return None	12×
217	return int(torstate.tor_pid)	12×
218
219	proc = subprocess.Popen(['lsof', '-i', '4tcp@%s:%s' % (addr, port)],	12×
220	stdout=subprocess.PIPE)
221	(stdout, stderr) = proc.communicate()	12×
222	lines = stdout.split(b'\n')	12×
223	if len(lines) > 1:	12×
224	return int(lines[1].split()[1])	12×
225
226
227	def hmac_sha256(key, msg):	12×
228	"""
229	Adapted from rransom's tor-utils git repository. Returns the
230	digest (binary) of an HMAC with SHA256 over msg with key.
231	"""
232
233	return hmac.new(key, msg, hashlib.sha256).digest()	12×
234
235
236	CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE = os.urandom(32)	12×
237
238
239	def compare_via_hash(x, y):	12×
240	"""
241	Taken from rransom's tor-utils git repository, to compare two
242	hashes in something resembling constant time (or at least, not
243	leaking timing info?)
244	"""
245	return (hmac_sha256(CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE, x) ==	12×
246	hmac_sha256(CRYPTOVARIABLE_EQUALITY_COMPARISON_NONCE, y))
247
248
249	class NetLocation(object):	12×
250	"""
251	Represents the location of an IP address, either city or country
252	level resolution depending on what GeoIP database was loaded. If
253	the ASN database is available you get that also.
254	"""
255
256	def __init__(self, ipaddr):	12×
257	"ipaddr should be a dotted-quad"
258	self.ip = ipaddr	12×
259	self.latlng = (None, None)	12×
260	self.countrycode = None	12×
261	self.city = None	12×
262	self.asn = None	12×
263
264	if self.ip is None or self.ip == 'unknown':	12×
265	return	12×
266
267	if city:	12×
268	try:	12×
269	r = city.record_by_addr(self.ip)	12×
270	except Exception:	12×
271	r = None	12×
272	if r is not None:	12×
273	self.countrycode = r['country_code']	12×
274	self.latlng = (r['latitude'], r['longitude'])	12×
275	try:	12×
276	self.city = (r['city'], r['region_code'])	12×
277	except KeyError:	12×
278	self.city = (r['city'], r['region_name'])	12×
279
280	elif country:	12×
281	self.countrycode = country.country_code_by_addr(ipaddr)	12×
282
283	else:
284	self.countrycode = ''	12×
285
286	if asn:	12×
287	try:	12×
288	self.asn = asn.org_by_addr(self.ip)	12×
289	except Exception:	12×
290	self.asn = None	12×
291
292
293	@implementer(IProtocolFactory)	12×
294	class NoOpProtocolFactory:
295	"""
296	This is an IProtocolFactory that does nothing. Used for testing,
297	and for :method:`available_tcp_port`
298	"""
299	def noop(self, args, *kw):	12×
300	pass	12×
301	buildProtocol = noop	12×
302	doStart = noop	12×
303	doStop = noop	12×
304
305
306	@defer.inlineCallbacks	12×
307	def available_tcp_port(reactor):
308	"""
309	Returns a Deferred firing an available TCP port on localhost.
310	It does so by listening on port 0; then stopListening and fires the
311	assigned port number.
312	"""
313
314	endpoint = serverFromString(reactor, 'tcp:0:interface=127.0.0.1')	12×
315	port = yield endpoint.listen(NoOpProtocolFactory())	12×
316	address = port.getHost()	12×
317	yield port.stopListening()	12×
318	defer.returnValue(address.port)	12×
319
320
321	def unescape_quoted_string(string):	12×
322	r'''
323	This function implementes the recommended functionality described in the
324	tor control-spec to be compatible with older tor versions:
325
326	* Read \\n \\t \\r and \\0 ... \\377 as C escapes.
327	* Treat a backslash followed by any other character as that character.
328
329	Except the legacy support for the escape sequences above this function
330	implements parsing of QuotedString using qcontent from
331
332	QuotedString = DQUOTE *qcontent DQUOTE
333
334	:param string: The escaped quoted string.
335	:returns: The unescaped string.
336	:raises ValueError: If the string is in a invalid form
337	(e.g. a single backslash)
338	'''
339	match = re.match(r'''^"((?:[^"\\]\|\\.)*)"$''', string)	12×
340	if not match:	12×
341	raise ValueError("Invalid quoted string", string)	12×
342	string = match.group(1)	12×
343	# remove backslash before all characters which should not be
344	# handeled as escape codes by string.decode('string-escape').
345	# This is needed so e.g. '\x00' is not unescaped as '\0'
346	string = re.sub(r'((?:^\|[^\\])(?:\\\\)*)\\([^ntr0-7\\])', r'\1\2', string)	12×
347	if six.PY3:	12×
348	# XXX hmmm?
UNCOV 349	return bytes(string, 'ascii').decode('unicode-escape')	6×
350	return string.decode('string-escape')	6×
351
352
353	def default_control_port():	12×
354	"""
355	This returns a default control port, which respects an environment
356	variable `TX_CONTROL_PORT`. Without the environment variable, this
357	returns 9151 (the Tor Browser Bundle default).
358
359	You shouldn't use this in "normal" code, this is a convenience for
360	the examples.
361	"""
362	try:	12×
363	return int(os.environ['TX_CONTROL_PORT'])	12×
364	except KeyError:	12×
365	return 9151	12×
366
367
368	class IListener(Interface):	12×
369	def add(callback):	12×
370	"""
371	Add a listener. The arguments to the callback are determined by whomever calls notify()
372	"""
373
374	def remove(callback):	12×
375	"""
376	Add a listener. The arguments to the callback are determined by whomever calls notify()
377	"""
378
379	def notify(args, *kw):	12×
380	"""
381	Calls every listener with the given args and keyword-args.
382
383	XXX errors? just log?
384	"""
385
386
387	def maybe_coroutine(obj):	12×
388	"""
389	If 'obj' is a coroutine and we're using Python3, wrap it in
390	ensureDeferred. Otherwise return the original object.
391
392	(This is to insert in all callback chains from user code, in case
393	that user code is Python3 and used 'async def')
394	"""
395	if six.PY3 and asyncio.iscoroutine(obj):	12×
UNCOV 396	return defer.ensureDeferred(obj)	3×
397	return obj	12×
398
399
400	@implementer(IListener)	12×
401	class _Listener(object):	12×
402	"""
403	Internal helper.
404	"""
405
406	def __init__(self):	12×
407	self._listeners = set()	12×
408
409	def add(self, callback):	12×
410	"""
411	Add a callback to this listener
412	"""
413	self._listeners.add(callback)	12×
414
415	__call__ = add #: alias for "add"	12×
416
417	def remove(self, callback):	12×
418	"""
419	Remove a callback from this listener
420	"""
421	self._listeners.remove(callback)	12×
422
423	def notify(self, args, *kw):	12×
424	"""
425	Calls all listeners with the specified args.
426
427	Returns a Deferred which callbacks when all the listeners
428	which return Deferreds have themselves completed.
429	"""
430	calls = []	12×
431
432	def failed(fail):	12×
433	# XXX use logger
434	fail.printTraceback()	12×
435
436	for cb in self._listeners:	12×
437	d = defer.maybeDeferred(cb, args, *kw)	12×
438	d.addCallback(maybe_coroutine)	12×
439	d.addErrback(failed)	12×
440	calls.append(d)	12×
441	return defer.DeferredList(calls)	12×
442
443
444	class _ListenerCollection(object):	12×
445	"""
446	Internal helper.
447
448	This collects all your valid event listeners together in one
449	object if you want.
450	"""
451	def __init__(self, valid_events):	12×
452	self._valid_events = valid_events	12×
453	for e in valid_events:	12×
454	setattr(self, e, _Listener())	12×
455
456	def __call__(self, event, callback):	12×
457	if event not in self._valid_events:	12×
458	raise Exception("Invalid event '{}'".format(event))	12×
459	getattr(self, event).add(callback)	12×
460
461	def remove(self, event, callback):	12×
462	if event not in self._valid_events:	12×
463	raise Exception("Invalid event '{}'".format(event))	12×
464	getattr(self, event).remove(callback)	12×
465
466	def notify(self, event, args, *kw):	12×
467	if event not in self._valid_events:	12×
468	raise Exception("Invalid event '{}'".format(event))	12×
469	getattr(self, event).notify(args, *kw)	12×
470
471
472	# similar to OneShotObserverList in Tahoe-LAFS
473	class SingleObserver(object):	12×
474	"""
475	A helper for ".when_*()" sort of functions.
476	"""
477	_NotFired = object()	12×
478
479	def __init__(self):	12×
480	self._observers = []	12×
481	self._fired = self._NotFired	12×
482
483	def when_fired(self):	12×
484	d = defer.Deferred()	12×
485	if self._fired is not self._NotFired:	12×
486	d.callback(self._fired)	12×
487	else:
488	self._observers.append(d)	12×
489	return d	12×
490
491	def fire(self, value):	12×
492	if self._observers is None:	12×
493	return # raise RuntimeError("already fired") ?	12×
494	self._fired = value	12×
495	for d in self._observers:	12×
496	d.callback(self._fired)	12×
497	self._observers = None	12×
498	return value # so we're transparent if used as a callback	12×
499
500
501	class _Version(object):	12×
502	"""
503	Replacement for incremental.Version until
504	https://github.com/meejah/txtorcon/issues/233 and/or
505	https://github.com/hawkowl/incremental/issues/31 is fixed.
506	"""
507	# as of latest incremental, it should only access .package and
508	# .short() via the getVersionString() method that Twisted's
509	# deprecated() uses...
510
511	def __init__(self, package, major, minor, patch):	12×
512	self.package = package	12×
513	self.major = major	12×
514	self.minor = minor	12×
515	self.patch = patch	12×
516
517	def short(self):	12×
518	return '{}.{}.{}'.format(self.major, self.minor, self.patch)	12×
519
520
521	# originally from magic-wormhole code
522	def _is_non_public_numeric_address(host):	12×
523	"""
524	returns True if 'host' is not public
525	"""
526	# for numeric hostnames, skip RFC1918 addresses, since no Tor exit
527	# node will be able to reach those. Likewise ignore IPv6 addresses.
528	try:	12×
529	a = ipaddress.ip_address(six.text_type(host))	12×
530	except ValueError:	12×
531	return False # non-numeric, let Tor try it	12×
532	if a.is_loopback or a.is_multicast or a.is_private or a.is_reserved \	12×
533	or a.is_unspecified:
534	return True # too weird, don't connect	12×
535	return False	12×

meejah / txtorcon / 1164

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous