[tor-commits] [stem/master] Helper function for resolution of inputs to fingerprints

commit fb6962ed28463a56d100c9cac7cc7517879dec99 Author: Damian Johnson atagar@torproject.org Date: Sun May 4 19:14:35 2014 -0700

Helper function for resolution of inputs to fingerprints

Splitting a big chunk of do_info() into a helper. Much better for both readability and testability. --- stem/interpretor/commands.py | 199 ++++++++++++++++++++----------------- stem/interpretor/help.py | 2 +- test/unit/interpretor/commands.py | 64 +++++++++++- 3 files changed, 170 insertions(+), 95 deletions(-)

diff --git a/stem/interpretor/commands.py b/stem/interpretor/commands.py index 2a9d402..faf9509 100644 --- a/stem/interpretor/commands.py +++ b/stem/interpretor/commands.py @@ -13,6 +13,70 @@ from stem.interpretor import STANDARD_OUTPUT, BOLD_OUTPUT, ERROR_OUTPUT, msg from stem.util.term import format

+def _get_fingerprint(arg, controller): + """ + Resolves user input into a relay fingerprint. This accepts... + + * Fingerprints + * Nicknames + * IPv4 addresses, either with or without an ORPort + * Empty input, which is resolved to ourselves if we're a relay + + :param str arg: input to be resolved to a relay fingerprint + :param stem.control.Controller controller: tor control connection + + :returns: **str** for the relay fingerprint + + :raises: **ValueError** if we're unable to resolve the input to a relay + """ + + if not arg: + try: + return controller.get_info('fingerprint') + except: + raise ValueError("We aren't a relay, no information to provide") + elif stem.util.tor_tools.is_valid_fingerprint(arg): + return arg + elif stem.util.tor_tools.is_valid_nickname(arg): + try: + return controller.get_network_status(arg).fingerprint + except: + raise ValueError("Unable to find a relay with the nickname of '%s'" % arg) + elif ':' in arg or stem.util.connection.is_valid_ipv4_address(arg): + if ':' in arg: + address, port = arg.split(':', 1) + + if not stem.util.connection.is_valid_ipv4_address(address): + raise ValueError("'%s' isn't a valid IPv4 address" % address) + elif port and not stem.util.connection.is_valid_port(port): + raise ValueError("'%s' isn't a valid port" % port) + + port = int(port) + else: + address, port = arg, None + + matches = {} + + for desc in controller.get_network_statuses(): + if desc.address == address: + if not port or desc.or_port == port: + matches[desc.or_port] = desc.fingerprint + + if len(matches) == 0: + raise ValueError('No relays found at %s' % arg) + elif len(matches) == 1: + return matches.values()[0] + else: + response = "There's multiple relays at %s, include a port to specify which.\n\n" % arg + + for i, or_port in enumerate(matches): + response += " %i. %s:%s, fingerprint: %s\n" % (i + 1, address, or_port, matches[or_port]) + + raise ValueError(response) + else: + raise ValueError("'%s' isn't a fingerprint, nickname, or IP address" % arg) + + class ControlInterpretor(object): """ Handles issuing requests and providing nicely formed responses, with support @@ -60,117 +124,66 @@ class ControlInterpretor(object): pretty fashion. """

- output, fingerprint = '', None + try: + fingerprint = _get_fingerprint(arg, self._controller) + except ValueError as exc: + return format(str(exc), *ERROR_OUTPUT)

- # determines the fingerprint, leaving it unset and adding an error message - # if unsuccessful + micro_desc = self._controller.get_microdescriptor(fingerprint, None) + server_desc = self._controller.get_server_descriptor(fingerprint, None) + ns_desc = self._controller.get_network_status(fingerprint, None)

- if not arg: - # uses our fingerprint if we're a relay, otherwise gives an error + # We'll mostly rely on the router status entry. Either the server + # descriptor or microdescriptor will be missing, so we'll treat them as + # being optional.

- fingerprint = self._controller.get_info('fingerprint', None) + if not ns_desc: + return format("Unable to find consensus information for %s" % fingerprint, *ERROR_OUTPUT)

- if not fingerprint: - output += format("We aren't a relay, no information to provide", *ERROR_OUTPUT) - elif stem.util.tor_tools.is_valid_fingerprint(arg): - fingerprint = arg - elif stem.util.tor_tools.is_valid_nickname(arg): - desc = self._controller.get_network_status(arg, None) + locale = self._controller.get_info('ip-to-country/%s' % ns_desc.address, None) + locale_label = ' (%s)' % locale if locale else ''

- if desc: - fingerprint = desc.fingerprint - else: - return format("Unable to find a relay with the nickname of '%s'" % arg, *ERROR_OUTPUT) - elif ':' in arg or stem.util.connection.is_valid_ipv4_address(arg): - # we got an address, so looking up the fingerprint - - if ':' in arg: - address, port = arg.split(':', 1) - - if not stem.util.connection.is_valid_ipv4_address(address): - return format("'%s' isn't a valid IPv4 address" % address, *ERROR_OUTPUT) - elif port and not stem.util.connection.is_valid_port(port): - return format("'%s' isn't a valid port" % port, *ERROR_OUTPUT) - - port = int(port) - else: - address, port = arg, None - - matches = {} - - for desc in self._controller.get_network_statuses(): - if desc.address == address: - if not port or desc.or_port == port: - matches[desc.or_port] = desc.fingerprint - - if len(matches) == 0: - output += format('No relays found at %s' % arg, *ERROR_OUTPUT) - elif len(matches) == 1: - fingerprint = matches.values()[0] - else: - output += format("There's multiple relays at %s, include a port to specify which.\n\n" % arg, *ERROR_OUTPUT) - - for i, or_port in enumerate(matches): - output += format(" %i. %s:%s, fingerprint: %s\n" % (i + 1, address, or_port, matches[or_port]), *ERROR_OUTPUT) + if server_desc: + exit_policy_label = server_desc.exit_policy.summary() + elif micro_desc: + exit_policy_label = micro_desc.exit_policy.summary() else: - return format("'%s' isn't a fingerprint, nickname, or IP address" % arg, *ERROR_OUTPUT) - - if fingerprint: - micro_desc = self._controller.get_microdescriptor(fingerprint, None) - server_desc = self._controller.get_server_descriptor(fingerprint, None) - ns_desc = self._controller.get_network_status(fingerprint, None) - - # We'll mostly rely on the router status entry. Either the server - # descriptor or microdescriptor will be missing, so we'll treat them as - # being optional. - - if not ns_desc: - return format("Unable to find consensus information for %s" % fingerprint, *ERROR_OUTPUT) - - locale = self._controller.get_info('ip-to-country/%s' % ns_desc.address, None) - locale_label = ' (%s)' % locale if locale else '' - - if server_desc: - exit_policy_label = server_desc.exit_policy.summary() - elif micro_desc: - exit_policy_label = micro_desc.exit_policy.summary() - else: - exit_policy_label = 'Unknown' + exit_policy_label = 'Unknown'

- output += '%s (%s)\n' % (ns_desc.nickname, fingerprint) + output = '%s (%s)\n' % (ns_desc.nickname, fingerprint)

- output += format('address: ', *BOLD_OUTPUT) - output += '%s:%s%s\n' % (ns_desc.address, ns_desc.or_port, locale_label) + output += format('address: ', *BOLD_OUTPUT) + output += '%s:%s%s\n' % (ns_desc.address, ns_desc.or_port, locale_label)

- output += format('published: ', *BOLD_OUTPUT) - output += ns_desc.published.strftime('%H:%M:%S %d/%m/%Y') + '\n' + output += format('published: ', *BOLD_OUTPUT) + output += ns_desc.published.strftime('%H:%M:%S %d/%m/%Y') + '\n'

- if server_desc: - output += format('os: ', *BOLD_OUTPUT) - output += server_desc.platform.decode('utf-8', 'replace') + '\n' + if server_desc: + output += format('os: ', *BOLD_OUTPUT) + output += server_desc.platform.decode('utf-8', 'replace') + '\n'

- output += format('version: ', *BOLD_OUTPUT) - output += str(server_desc.tor_version) + '\n' + output += format('version: ', *BOLD_OUTPUT) + output += str(server_desc.tor_version) + '\n'

- output += format('flags: ', *BOLD_OUTPUT) - output += ', '.join(ns_desc.flags) + '\n' + output += format('flags: ', *BOLD_OUTPUT) + output += ', '.join(ns_desc.flags) + '\n'

- output += format('exit policy: ', *BOLD_OUTPUT) - output += exit_policy_label + '\n' + output += format('exit policy: ', *BOLD_OUTPUT) + output += exit_policy_label + '\n'

- if server_desc: - contact = server_desc.contact + if server_desc: + contact = server_desc.contact

- # clears up some highly common obscuring + # clears up some highly common obscuring

- for alias in (' at ', ' AT '): - contact = contact.replace(alias, '@') + for alias in (' at ', ' AT '): + contact = contact.replace(alias, '@')

- for alias in (' dot ', ' DOT '): - contact = contact.replace(alias, '.') + for alias in (' dot ', ' DOT '): + contact = contact.replace(alias, '.')

- output += format('contact: ', *BOLD_OUTPUT) - output += contact + '\n' + output += format('contact: ', *BOLD_OUTPUT) + output += contact + '\n'

return output.strip()

diff --git a/stem/interpretor/help.py b/stem/interpretor/help.py index 20439d6..039bbef 100644 --- a/stem/interpretor/help.py +++ b/stem/interpretor/help.py @@ -23,7 +23,7 @@ def response(controller, arg): """ Provides our /help response.

- :param stem.Controller controller: tor control connection + :param stem.control.Controller controller: tor control connection :param str arg: controller or interpretor command to provide help output for

:returns: **str** with our help response diff --git a/test/unit/interpretor/commands.py b/test/unit/interpretor/commands.py index db0a53b..1517e3a 100644 --- a/test/unit/interpretor/commands.py +++ b/test/unit/interpretor/commands.py @@ -1,20 +1,82 @@ import unittest

+import stem import stem.response

-from stem.interpretor.commands import ControlInterpretor +from stem.interpretor.commands import ControlInterpretor, _get_fingerprint

from test import mocking from test.unit.interpretor import CONTROLLER

+try: + # added in python 3.3 + from unittest.mock import Mock +except ImportError: + from mock import Mock + EXPECTED_EVENTS_RESPONSE = """\ \x1b[34mBW 15 25\x1b[0m \x1b[34mBW 758 570\x1b[0m \x1b[34mDEBUG connection_edge_process_relay_cell(): Got an extended cell! Yay.\x1b[0m """

+FINGERPRINT = '9695DFC35FFEB861329B9F1AB04C46397020CE31' +

class TestInterpretorCommands(unittest.TestCase): + def test_get_fingerprint_for_ourselves(self): + controller = Mock() + + controller.get_info.side_effect = lambda arg: { + 'fingerprint': FINGERPRINT, + }[arg] + + self.assertEqual(FINGERPRINT, _get_fingerprint('', controller)) + + controller.get_info.side_effect = stem.ControllerError + self.assertRaises(ValueError, _get_fingerprint, '', controller) + + def test_get_fingerprint_for_fingerprint(self): + self.assertEqual(FINGERPRINT, _get_fingerprint(FINGERPRINT, Mock())) + + def test_get_fingerprint_for_nickname(self): + controller, descriptor = Mock(), Mock() + descriptor.fingerprint = FINGERPRINT + + controller.get_network_status.side_effect = lambda arg: { + 'moria1': descriptor, + }[arg] + + self.assertEqual(FINGERPRINT, _get_fingerprint('moria1', controller)) + + controller.get_network_status.side_effect = stem.ControllerError + self.assertRaises(ValueError, _get_fingerprint, 'moria1', controller) + + def test_get_fingerprint_for_address(self): + controller = Mock() + + self.assertRaises(ValueError, _get_fingerprint, '127.0.0.1:-1', controller) + self.assertRaises(ValueError, _get_fingerprint, '127.0.0.901:80', controller) + + descriptor = Mock() + descriptor.address = '127.0.0.1' + descriptor.or_port = 80 + descriptor.fingerprint = FINGERPRINT + + controller.get_network_statuses.return_value = [descriptor] + + self.assertEqual(FINGERPRINT, _get_fingerprint('127.0.0.1', controller)) + self.assertEqual(FINGERPRINT, _get_fingerprint('127.0.0.1:80', controller)) + self.assertRaises(ValueError, _get_fingerprint, '127.0.0.1:81', controller) + self.assertRaises(ValueError, _get_fingerprint, '127.0.0.2', controller) + + def test_get_fingerprint_for_unrecognized_inputs(self): + self.assertRaises(ValueError, _get_fingerprint, 'blarg!', Mock()) + + def test_quit(self): + interpretor = ControlInterpretor(CONTROLLER) + self.assertRaises(stem.SocketClosed, interpretor.run_command, '/quit') + def test_help(self): interpretor = ControlInterpretor(CONTROLLER)