Add CLI interface, fix bug in TLS validation, add logging

.pylintrc

@@ -1,2 +1,3 @@
 [TYPECHECK]
 generated-members = listenTCP, addCallback, _wrapContextFactory, _policyForHTTPS
+ignored-classes=twisted.internet.reactor

bwscanner/attacher.py

@@ -1,11 +1,22 @@
 import sys
+import itertools
+
+from twisted.python import log
 from twisted.internet import defer, reactor
 from txtorcon.interface import CircuitListenerMixin, IStreamAttacher, StreamListenerMixin
 from txtorcon import TorState, launch_tor
 from txtorcon.util import available_tcp_port
 from zope.interface import implementer
 
 
+FETCH_ALL_DESCRIPTOR_OPTIONS = {
+    'UseMicroDescriptors': 0,
+    'FetchUselessDescriptors': 1,
+    'FetchDirInfoEarly': 1,
+    'FetchDirInfoExtraEarly': 1,
+}
+
+
 @implementer(IStreamAttacher)
 class SOCKSClientStreamAttacher(CircuitListenerMixin, StreamListenerMixin):
     """
@@ -130,11 +141,12 @@ def launch_and_get_state(ignore):
     return get_random_tor_ports().addCallback(launch_and_get_state)
 
 
-def setconf_fetch_all_descs(tor):
-    # ensure that we have server descriptors available
-    d = tor.protocol.set_conf('UseMicroDescriptors', '0',
-                              'FetchUselessDescriptors', '1',
-                              'FetchDirInfoEarly', '1')
+def update_tor_config(tor, config):
+    """
+    Update the Tor config from a dict of config key: value pairs.
+    """
+    config_pairs = [(key, value) for key, value in config.items()]
+    d = tor.protocol.set_conf(*itertools.chain.from_iterable(config_pairs))
     return d.addCallback(lambda result: tor)
 
 

bwscanner/fetcher.py

@@ -1,6 +1,6 @@
 from twisted.internet import interfaces, reactor
 from twisted.internet.endpoints import TCP4ClientEndpoint
-from twisted.web.client import SchemeNotSupported, Agent
+from twisted.web.client import SchemeNotSupported, Agent, BrowserLikePolicyForHTTPS
 from txsocksx.client import SOCKS5ClientFactory
 from txsocksx.tls import TLSWrapClientEndpoint
 from zope.interface import implementer
@@ -62,6 +62,7 @@ def _create_circ(proto):
 
 class OnionRoutedAgent(Agent):
     _tlsWrapper = TLSWrapClientEndpoint
+    _policyForHTTPS = BrowserLikePolicyForHTTPS
 
     def __init__(self, *args, **kw):
         self.path = kw.pop('path')
@@ -82,9 +83,8 @@ def _getEndpoint(self, parsedURI, host=None, port=None):
             if hasattr(self, '_wrapContextFactory'):
                 tls_policy = self._wrapContextFactory(host, port)
             elif hasattr(self, '_policyForHTTPS'):
-                tls_policy = self._policyForHTTPS.creatorForNetloc(host, port)
+                tls_policy = self._policyForHTTPS().creatorForNetloc(host, port)
             else:
-                raise NotImplementedError(
-                    "can't figure out how to make a context factory")
+                raise NotImplementedError("Cannot create a TLS validation policy.")
             endpoint = self._tlsWrapper(tls_policy, endpoint)
         return endpoint

bwscanner/measurement.py

@@ -12,6 +12,8 @@
 from bwscanner.fetcher import OnionRoutedAgent
 from bwscanner.writer import ResultSink
 
+# defer.setDebugging(True)
+
 
 class DownloadIncomplete(Exception):
     pass
@@ -35,7 +37,7 @@ def __init__(self, state, clock, log_dir, **kwargs):
         self.partitions = kwargs.get('partitions', 1)
         self.this_partition = kwargs.get('this_partition', 0)
         self.scan_continuous = kwargs.get('scan_continuous', False)
-        self.circuit_lifetime = kwargs.get('circuit_lifetime', 60)
+        self.request_timeout = kwargs.get('request_timeout', 60)
         self.circuit_launch_delay = kwargs.get('circuit_launch_delay', .2)
 
         self.tasks = []
@@ -116,6 +118,7 @@ def get_circuit_bw(result):
                 report['path_desc_bws'].append((yield self.get_r_desc_bw(relay)))
                 report['path_ns_bws'].append((yield self.get_r_ns_bw(relay)))
             report['path_bws'] = [r.bandwidth for r in path]
+            log.msg("Download successful for router %s." % path[0].id_hex)
             defer.returnValue(report)
 
         def circ_failure(failure):
@@ -125,17 +128,30 @@ def circ_failure(failure):
             report['time_start'] = time_start
             report['path'] = [r.id_hex for r in path]
             report['failure'] = failure.__repr__()
+            log.msg("Download failed for router %s: %s." % (path[0].id_hex, report['failure']))
             return report
 
+        def timeoutDeferred(deferred, timeout):
+            def cancelDeferred(deferred):
+                log.msg("Cancelling deferred", deferred)
+                deferred.cancel()
+
+            delayedCall = self.clock.callLater(timeout, cancelDeferred, deferred)
+
+            def gotResult(result):
+                if delayedCall.active():
+                    log.msg("Cancel timeout")
+                    delayedCall.cancel()
+                return result
+            deferred.addBoth(gotResult)
+
         agent = OnionRoutedAgent(self.clock, path=path, state=self.state)
         request = agent.request("GET", url)
-        timeout_circuit = self.clock.callLater(self.circuit_lifetime, request.cancel)
-        request.addCallback(readBody)
-        request.addCallbacks(get_circuit_bw, errback=circ_failure)
+        request.addCallback(readBody)  # returns a readBody Deferred
+        timeoutDeferred(request, self.request_timeout)
+        request.addCallbacks(get_circuit_bw)
+        request.addErrback(circ_failure)
         request.addCallback(self.result_sink.send)
-
-        # Stop circuit timeout callLater when we have been successful
-        request.addCallback(lambda _: timeout_circuit.cancel())
         self.tasks.append(request)
 
     @defer.inlineCallbacks

bwscanner/scanner.py

@@ -0,0 +1,123 @@
+import os
+import sys
+
+import click
+from twisted.internet import reactor
+from twisted.python import log
+from txtorcon import build_local_tor_connection, TorConfig
+
+from bwscanner.attacher import start_tor, update_tor_config, FETCH_ALL_DESCRIPTOR_OPTIONS
+from bwscanner.measurement import BwScan
+
+
+BWSCAN_VERSION = '0.0.1'
+
+
+class ScanInstance(object):
+    """
+    Store the configuration and state for the CLI tool.
+    """
+    def __init__(self, data_dir):
+        self.data_dir = data_dir
+        self.verbose = False
+
+    def __repr__(self):
+        return '<BWScan %r>' % self.data_dir
+
+pass_scan = click.make_pass_decorator(ScanInstance)
+
+
+@click.group()
+@click.option('--verbose/--no-verbose', default=False,
+              help='Enables verbose mode.')
+@click.option('--data-dir', type=click.Path(),
+              default=os.environ.get("BWSCANNER_DATADIR", click.get_app_dir('bwscanner')),
+              help='Directory where bwscan should stores its measurements and '
+              'other data.')
+@click.version_option(BWSCAN_VERSION)
+@click.pass_context
+def cli(ctx, verbose, data_dir):
+    """
+    The bwscan tool measures Tor relays and calculates their bandwidth. These
+    bandwidth measurements can then be aggregate to create the bandwidth
+    values used by the Tor bandwidth authorities when creating the Tor consensus.
+    """
+    # Create the data directory if it doesn't exist
+    data_dir = os.path.abspath(data_dir)
+    if not os.path.exists(data_dir):
+        os.makedirs(data_dir)
+    ctx.obj = ScanInstance(data_dir)
+    ctx.obj.verbose = verbose
+
+
+@cli.command(short_help="Measure the Tor relays.")
+@click.option('--partitions', '-p', default=1,
+              help='Divide the set of relays into subsets. 1 by default.')
+@click.option('--current-partition', '-c', default=1,
+              help='Scan a particular subset / partition of the relays.')
+@click.option('--timeout', default=120,
+              help='Timeout for measurement HTTP requests (default: %ds).' % 120)
+@click.option('--launch-tor/--no-launch-tor', default=True,
+              help='Launch Tor or try to connect to an existing Tor instance.')
+@click.option('--circuit-build-timeout', default=20,
+              help='Option passed when launching Tor.')
+@click.option('--circuit-idle-timeout', default=20,
+              help='Option passed when launching Tor.')
+@pass_scan
+def scan(scan, launch_tor, partitions, current_partition, timeout,
+         circuit_build_timeout, circuit_idle_timeout):
+    """
+    Start a scan through each Tor relay to measure it's bandwidth.
+    """
+    if scan.verbose:
+        log.startLogging(sys.stdout)
+        click.echo('Verbose log mode is on.')
+        click.echo('Using %s as the data directory.' % scan.data_dir)
+
+    # Options for spawned or running Tor to load the correct descriptors.
+    tor_options = {
+        'LearnCircuitBuildTimeout': 0,  # Disable adaptive circuit timeouts.
+        'CircuitBuildTimeout': circuit_build_timeout,
+        'CircuitIdleTimeout': circuit_idle_timeout,
+    }
+
+    def tor_status(tor):
+        click.echo("Connected to a Tor instance.")
+        return tor
+
+    if launch_tor:
+        click.echo("Spawning a new Tor instance")
+        c = TorConfig()
+        # Update Tor config before launching a new Tor.
+        c.config.update(tor_options)
+        c.config.update(FETCH_ALL_DESCRIPTOR_OPTIONS)
+        tor = start_tor(c)
+
+    else:
+        click.echo("Connecting to a running Tor instance")
+        tor = build_local_tor_connection(reactor)
+        # Update the Tor config on a running Tor.
+        tor.addCallback(update_tor_config, tor_options)
+        tor.addCallback(update_tor_config, FETCH_ALL_DESCRIPTOR_OPTIONS)
+
+    tor.addCallback(tor_status)
+
+    # XXX: check that each run is producing the same input set!
+    measurement_dir = os.path.join(scan.data_dir, 'measurements')
+    if not os.path.exists(measurement_dir):
+        os.makedirs(measurement_dir)
+
+    tor.addCallback(BwScan, reactor, measurement_dir, request_timeout=timeout,
+                    partitions=partitions, this_partition=current_partition)
+    tor.addCallback(lambda scanner: scanner.run_scan())
+    tor.addCallback(lambda _: reactor.stop())
+    reactor.run()
+
+
+@cli.command(short_help="Combine bandwidth measurements.")
+def aggregate():
+    """
+    Command to aggregate BW measurements to create file for the BWAuths
+    """
+    click.echo('Aggregating bandwidth measurements')
+    click.echo('Not implemented yet')

setup.py

@@ -23,5 +23,9 @@
       license=__license__,
       packages=find_packages(),
       # data_files = [('path', ['filename'])]
-      data_files=[]
+      data_files=[],
+      entry_points={
+        "console_scripts": [
+            'bwscan = bwscanner.scanner:cli',
+        ]},
      )

test/test_measurement.py

@@ -2,7 +2,7 @@
 from twisted.web.resource import Resource
 from twisted.web.server import Site
 from txtorcon.util import available_tcp_port
-from bwscanner.attacher import setconf_fetch_all_descs
+from bwscanner.attacher import update_tor_config, FETCH_ALL_DESCRIPTOR_OPTIONS
 from bwscanner.measurement import BwScan
 from test.template import TorTestCase
 from tempfile import mkdtemp
@@ -18,7 +18,7 @@ class TestBwscan(TorTestCase):
     @defer.inlineCallbacks
     def setUp(self):
         yield super(TestBwscan, self).setUp()
-        yield setconf_fetch_all_descs(self.tor)
+        yield update_tor_config(self.tor, FETCH_ALL_DESCRIPTOR_OPTIONS)
 
         class DummyResource(Resource):
             isLeaf = True