[tor-commits] [stem/master] Support for ephemeral hidden services
atagar at torproject.org
atagar at torproject.org
Sun May 10 20:38:47 UTC 2015
commit 880b34d6b0acb7535fcff4fe322ebf4a83f81e5d
Author: Damian Johnson <atagar at torproject.org>
Date: Sun May 10 13:32:58 2015 -0700
Support for ephemeral hidden services
Adding support for Yawning's new style for hidden services that don't touch
disk...
https://gitweb.torproject.org/torspec.git/commit/?id=f5ff369
These are far easier to work with since they sidestep the headaches of reading
the address and private key from disk. Hopefully in the long term these will
replace the traditional method of making hidden services so we can merge our
create_hidden_service() and create_ephemeral_hidden_service() methods.
---
docs/change_log.rst | 3 +
stem/control.py | 157 ++++++++++++++++++++++++++++++++++++++
stem/interpreter/settings.cfg | 6 ++
stem/response/__init__.py | 9 ++-
stem/response/add_onion.py | 43 +++++++++++
stem/util/log.py | 3 +
stem/version.py | 4 +
test/integ/control/controller.py | 145 +++++++++++++++++++++++++++++++++++
test/settings.cfg | 1 +
test/unit/response/add_onion.py | 95 +++++++++++++++++++++++
10 files changed, 463 insertions(+), 3 deletions(-)
diff --git a/docs/change_log.rst b/docs/change_log.rst
index 4e2bf98..5aa0ab1 100644
--- a/docs/change_log.rst
+++ b/docs/change_log.rst
@@ -51,6 +51,7 @@ conversion (:trac:`14075`).
* **Controller**
+ * Added :class:`~stem.control.Controller` methods for a new style of hidden services that don't touch disk: :func:`~stem.control.Controller.list_ephemeral_hidden_services`, :func:`~stem.control.Controller.create_ephemeral_hidden_service`, and :func:`~stem.control.Controller.remove_ephemeral_hidden_service` (:spec:`f5ff369`)
* Added :func:`~stem.control.Controller.get_hidden_service_descriptor` and `support for HS_DESC_CONTENT events <api/response.html#stem.response.events.HSDescContentEvent>`_ (:trac:`14847`, :spec:`aaf2434`)
* :func:`~stem.process.launch_tor_with_config` avoids writing a temporary torrc to disk if able (:trac:`13865`)
* :class:`~stem.response.events.CircuitEvent` support for the new SOCKS_USERNAME and SOCKS_PASSWORD arguments (:trac:`14555`, :spec:`2975974`)
@@ -79,6 +80,8 @@ conversion (:trac:`14075`).
* **Website**
* Added support and `instructions for tox <faq.html#how-do-i-test-compatibility-with-multiple-python-versions>`_ (:trac:`14091`)
+ * Added OSX to our `download page <download.html>`_ (:trac:`8588`)
+ * Updated our twitter example to work with the service's 1.1 API (:trac:`9003`)
.. _version_1.3:
diff --git a/stem/control.py b/stem/control.py
index 48171d0..3e426ea 100644
--- a/stem/control.py
+++ b/stem/control.py
@@ -101,6 +101,10 @@ If you're fine with allowing your script to raise exceptions then this can be mo
|- create_hidden_service - creates a new hidden service or adds a new port
|- remove_hidden_service - removes a hidden service or drops a port
|
+ |- list_ephemeral_hidden_services - list ephemeral hidden serivces
+ |- create_ephemeral_hidden_service - create a new ephemeral hidden service
+ |- remove_ephemeral_hidden_service - removes an ephemeral hidden service
+ |
|- add_event_listener - attaches an event listener to be notified of tor events
|- remove_event_listener - removes a listener so it isn't notified of further events
|
@@ -2395,6 +2399,10 @@ class Controller(BaseController):
However, this directory is only readable by the tor user, so if unavailable
the **hostname** will be **None**.
+ **As of Tor 0.2.7.1 there's two ways for creating hidden services. This is
+ no longer the recommended method.** Rather, try using
+ :func:`~stem.control.Controller.create_ephemeral_hidden_service` instead.
+
.. versionadded:: 1.3.0
.. versionchanged:: 1.4.0
@@ -2529,6 +2537,155 @@ class Controller(BaseController):
self.set_hidden_service_conf(conf)
return True
+ @with_default()
+ def list_ephemeral_hidden_services(self, default = UNDEFINED, our_services = True, detached = False):
+ """
+ list_ephemeral_hidden_services(default = UNDEFINED, our_services = True, detached = False)
+
+ Lists hidden service addresses created by
+ :func:`~stem.control.Controller.create_ephemeral_hidden_service`.
+
+ .. versionadded:: 1.4.0
+
+ :param object default: response if the query fails
+ :param bool our_services: include services created with this controller
+ that weren't flagged as 'detached'
+ :param bool detached: include services whos contiuation isn't tied to a
+ controller
+
+ :returns: **list** of hidden service addresses without their '.onion'
+ suffix
+
+ :raises: :class:`stem.ControllerError` if the call fails and we weren't
+ provided a default response
+ """
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # if self.get_version() < stem.version.Requirement.ADD_ONION:
+ # raise stem.UnsatisfiableRequest(message = 'Ephemeral hidden services were added in tor version %s' % stem.version.Requirement.ADD_ONION)
+
+ result = []
+
+ if our_services:
+ try:
+ result += self.get_info('onions/current').split('\n')
+ except stem.ProtocolError as exc:
+ if 'No onion services of the specified type.' not in str(exc):
+ raise exc
+
+ if detached:
+ try:
+ result += self.get_info('onions/detached').split('\n')
+ except stem.ProtocolError as exc:
+ if 'No onion services of the specified type.' not in str(exc):
+ raise exc
+
+ return result
+
+ def create_ephemeral_hidden_service(self, ports, key_type = 'NEW', key_content = 'BEST', discard_key = False, detached = False):
+ """
+ Creates a new hidden service. Unlike
+ :func:`~stem.control.Controller.create_hidden_service` this style of
+ hidden service doesn't touch disk, carrying with it a lot of advantages.
+ This is the suggested method for making hidden services.
+
+ Our **ports** argument can be a single port...
+
+ ::
+
+ create_ephemeral_hidden_service(80)
+
+ ... list of ports the service is available on...
+
+ ::
+
+ create_ephemeral_hidden_service([80, 443])
+
+ ... or a mapping of hidden service ports to their targets...
+
+ ::
+
+ create_ephemeral_hidden_service({80: 80, 443: '173.194.33.133:443'})
+
+ .. versionadded:: 1.4.0
+
+ :param int,list,dict ports: hidden service port(s) or mapping of hidden
+ service ports to their targets
+ :param str key_type: type of key being provided, generates a new key if
+ 'NEW' (options are: **NEW** and **RSA1024**)
+ :param str key_content: key for the service to use or type of key to be
+ generated (options when **key_type** is **NEW** are **BEST** and
+ **RSA1024**)
+ :param bool discard_key: avoid providing the key back in our response
+ :param bool detached: continue this hidden service even after this control
+ connection is closed if **True**
+
+ :returns: :class:`~stem.response.AddOnionResponse` with the response
+
+ :raises: :class:`stem.ControllerError` if the call fails
+ """
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # if self.get_version() < stem.version.Requirement.ADD_ONION:
+ # raise stem.UnsatisfiableRequest(message = 'Ephemeral hidden services were added in tor version %s' % stem.version.Requirement.ADD_ONION)
+
+ request = 'ADD_ONION %s:%s' % (key_type, key_content)
+
+ flags = []
+
+ if discard_key:
+ flags.append('DiscardPK')
+
+ if detached:
+ flags.append('Detach')
+
+ if flags:
+ request += ' Flags=%s' % ','.join(flags)
+
+ if isinstance(ports, int):
+ request += ' Port=%s' % ports
+ elif isinstance(ports, list):
+ for port in ports:
+ request += ' Port=%s' % port
+ elif isinstance(ports, dict):
+ for port, target in ports.items():
+ request += ' Port=%s,%s' % (port, target)
+ else:
+ raise ValueError("The 'ports' argument of create_ephemeral_hidden_service() needs to be an int, list, or dict")
+
+ response = self.msg(request)
+ stem.response.convert('ADD_ONION', response)
+ return response
+
+ def remove_ephemeral_hidden_service(self, service_id):
+ """
+ Discontinues a given hidden service that was created with
+ :func:`~stem.control.Controller.create_ephemeral_hidden_service`.
+
+ .. versionadded:: 1.4.0
+
+ :param str service_id: hidden service address without the '.onion' suffix
+
+ :returns: **True** if the hidden service is discontinued, **False** if it
+ wasn't running in the first place
+
+ :raises: :class:`stem.ControllerError` if the call fails
+ """
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # if self.get_version() < stem.version.Requirement.ADD_ONION:
+ # raise stem.UnsatisfiableRequest(message = 'Ephemeral hidden services were added in tor version %s' % stem.version.Requirement.ADD_ONION)
+
+ response = self.msg('DEL_ONION %s' % service_id)
+ stem.response.convert('SINGLELINE', response)
+
+ if response.is_ok():
+ return True
+ elif response.code == '552':
+ return False # no hidden service to discontinue
+ else:
+ raise stem.ProtocolError('DEL_ONION returned unexpected response code: %s' % response.code)
+
def add_event_listener(self, listener, *events):
"""
Directs further tor controller events to a given function. The function is
diff --git a/stem/interpreter/settings.cfg b/stem/interpreter/settings.cfg
index e026f6f..1bacf1c 100644
--- a/stem/interpreter/settings.cfg
+++ b/stem/interpreter/settings.cfg
@@ -317,4 +317,10 @@ autocomplete PROTOCOLINFO
autocomplete TAKEOWNERSHIP
autocomplete AUTHCHALLENGE
autocomplete DROPGUARDS
+autocomplete ADD_ONION NEW:BEST
+autocomplete ADD_ONION NEW:RSA1024
+autocomplete ADD_ONION RSA1024:
+autocomplete DEL_ONION
+autocomplete HSFETCH
+autocomplete HSPOST
diff --git a/stem/response/__init__.py b/stem/response/__init__.py
index 09b7de7..df534a9 100644
--- a/stem/response/__init__.py
+++ b/stem/response/__init__.py
@@ -31,6 +31,7 @@ Parses replies from the control socket.
"""
__all__ = [
+ 'add_onion',
'events',
'getinfo',
'getconf',
@@ -103,24 +104,26 @@ def convert(response_type, message, **kwargs):
or response_type isn't supported
"""
+ import stem.response.add_onion
+ import stem.response.authchallenge
import stem.response.events
import stem.response.getinfo
import stem.response.getconf
- import stem.response.protocolinfo
- import stem.response.authchallenge
import stem.response.mapaddress
+ import stem.response.protocolinfo
if not isinstance(message, ControlMessage):
raise TypeError('Only able to convert stem.response.ControlMessage instances')
response_types = {
+ 'ADD_ONION': stem.response.add_onion.AddOnionResponse,
+ 'AUTHCHALLENGE': stem.response.authchallenge.AuthChallengeResponse,
'EVENT': stem.response.events.Event,
'GETINFO': stem.response.getinfo.GetInfoResponse,
'GETCONF': stem.response.getconf.GetConfResponse,
'MAPADDRESS': stem.response.mapaddress.MapAddressResponse,
'SINGLELINE': SingleLineResponse,
'PROTOCOLINFO': stem.response.protocolinfo.ProtocolInfoResponse,
- 'AUTHCHALLENGE': stem.response.authchallenge.AuthChallengeResponse,
}
try:
diff --git a/stem/response/add_onion.py b/stem/response/add_onion.py
new file mode 100644
index 0000000..1472668
--- /dev/null
+++ b/stem/response/add_onion.py
@@ -0,0 +1,43 @@
+# Copyright 2015, Damian Johnson and The Tor Project
+# See LICENSE for licensing information
+
+import stem.response
+
+
+class AddOnionResponse(stem.response.ControlMessage):
+ """
+ ADD_ONION response.
+
+ :var str service_id: hidden service address without the '.onion' suffix
+ :var str private_key: base64 encoded hidden service private key
+ :var str private_key_type: crypto used to generate the hidden service private
+ key (such as RSA1024)
+ """
+
+ def _parse_message(self):
+ # Example:
+ # 250-ServiceID=gfzprpioee3hoppz
+ # 250-PrivateKey=RSA1024:MIICXgIBAAKBgQDZvYVxv...
+ # 250 OK
+
+ self.service_id = None
+ self.private_key = None
+ self.private_key_type = None
+
+ if not self.is_ok():
+ raise stem.ProtocolError("ADD_ONION response didn't have an OK status: %s" % self)
+
+ if not str(self).startswith('ServiceID='):
+ raise stem.ProtocolError('ADD_ONION response should start with the service id: %s' % self)
+
+ for line in list(self):
+ if '=' in line:
+ key, value = line.split('=', 1)
+
+ if key == 'ServiceID':
+ self.service_id = value
+ elif key == 'PrivateKey':
+ if ':' not in value:
+ raise stem.ProtocolError("ADD_ONION PrivateKey lines should be of the form 'PrivateKey=[type]:[key]: %s" % self)
+
+ self.private_key_type, self.private_key = value.split(':', 1)
diff --git a/stem/util/log.py b/stem/util/log.py
index 85bcd0f..4154706 100644
--- a/stem/util/log.py
+++ b/stem/util/log.py
@@ -198,6 +198,9 @@ class LogBuffer(logging.Handler):
"""
Basic log handler that listens for stem events and stores them so they can be
read later. Log entries are cleared as they are read.
+
+ .. versionchanged:: 1.4.0
+ Added the yield_records argument.
"""
def __init__(self, runlevel, yield_records = False):
diff --git a/stem/version.py b/stem/version.py
index 2289a3d..1182bfc 100644
--- a/stem/version.py
+++ b/stem/version.py
@@ -55,6 +55,8 @@ easily parsed and compared, for instance...
**FEATURE_VERBOSE_NAMES** 'VERBOSE_NAMES' optional feature
**GETINFO_CONFIG_TEXT** 'GETINFO config-text' query
**HSFETCH** HSFETCH requests
+ **HSPOST** HSPOST requests
+ **ADD_ONION** ADD_ONION and DEL_ONION requests
**LOADCONF** LOADCONF requests
**MICRODESCRIPTOR_IS_DEFAULT** Tor gets microdescriptors by default rather than server descriptors
**TAKEOWNERSHIP** TAKEOWNERSHIP requests
@@ -362,6 +364,8 @@ Requirement = stem.util.enum.Enum(
('FEATURE_VERBOSE_NAMES', Version('0.2.2.1-alpha')),
('GETINFO_CONFIG_TEXT', Version('0.2.2.7-alpha')),
('HSFETCH', Version('0.2.7.1-alpha')),
+ ('HSPOST', Version('0.2.7.1-alpha')),
+ ('ADD_ONION', Version('0.2.7.1-alpha')),
('LOADCONF', Version('0.2.1.1')),
('MICRODESCRIPTOR_IS_DEFAULT', Version('0.2.3.3')),
('TAKEOWNERSHIP', Version('0.2.2.28-beta')),
diff --git a/test/integ/control/controller.py b/test/integ/control/controller.py
index 01480ba..2557dfe 100644
--- a/test/integ/control/controller.py
+++ b/test/integ/control/controller.py
@@ -578,6 +578,151 @@ class TestController(unittest.TestCase):
except:
pass
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # @require_version(Requirement.ADD_ONION)
+
+ @require_controller
+ def test_without_ephemeral_hidden_services(self):
+ """
+ Exercises ephemeral hidden service methods when none are present.
+ """
+
+ with test.runner.get_runner().get_tor_controller() as controller:
+ self.assertEqual([], controller.list_ephemeral_hidden_services())
+ self.assertEqual([], controller.list_ephemeral_hidden_services(detached = True))
+ self.assertEqual(False, controller.remove_ephemeral_hidden_service('gfzprpioee3hoppz'))
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # @require_version(Requirement.ADD_ONION)
+
+ @require_controller
+ def test_with_ephemeral_hidden_services(self):
+ """
+ Exercises creating ephemeral hidden services and methods when they're
+ present.
+ """
+
+ runner = test.runner.get_runner()
+
+ with runner.get_tor_controller() as controller:
+ for ports in (4567890, [4567, 4567890], {4567: 'not_an_address:4567'}):
+ try:
+ # try creating a service with an invalid port
+ response = controller.create_ephemeral_hidden_service(ports)
+ self.fail("we should've raised a stem.ProtocolError")
+ except stem.ProtocolError as exc:
+ self.assertEqual("ADD_ONION response didn't have an OK status: Invalid VIRTPORT/TARGET", str(exc))
+
+ response = controller.create_ephemeral_hidden_service(4567)
+ self.assertEqual([response.service_id], controller.list_ephemeral_hidden_services())
+ self.assertTrue(response.private_key is not None)
+
+ # drop the service
+
+ self.assertEqual(True, controller.remove_ephemeral_hidden_service(response.service_id))
+ self.assertEqual([], controller.list_ephemeral_hidden_services())
+
+ # recreate the service with the same private key
+
+ recreate_response = controller.create_ephemeral_hidden_service(4567, key_type = response.private_key_type, key_content = response.private_key)
+ self.assertEqual([response.service_id], controller.list_ephemeral_hidden_services())
+ self.assertEqual(response.service_id, recreate_response.service_id)
+
+ # the response only includes the private key when making a new one
+
+ self.assertEqual(None, recreate_response.private_key)
+ self.assertEqual(None, recreate_response.private_key_type)
+
+ # create a service where we never see the private key
+
+ response = controller.create_ephemeral_hidden_service(4568, discard_key = True)
+ self.assertTrue(response.service_id in controller.list_ephemeral_hidden_services())
+ self.assertEqual(None, response.private_key)
+ self.assertEqual(None, response.private_key_type)
+
+ # other controllers shouldn't be able to see these hidden services
+
+ with runner.get_tor_controller() as second_controller:
+ self.assertEqual(2, len(controller.list_ephemeral_hidden_services()))
+ self.assertEqual(0, len(second_controller.list_ephemeral_hidden_services()))
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # @require_version(Requirement.ADD_ONION)
+
+ @require_controller
+ def test_with_detached_ephemeral_hidden_services(self):
+ """
+ Exercises creating detached ephemeral hidden services and methods when
+ they're present.
+ """
+
+ runner = test.runner.get_runner()
+
+ with runner.get_tor_controller() as controller:
+ response = controller.create_ephemeral_hidden_service(4567, detached = True)
+ self.assertEqual([], controller.list_ephemeral_hidden_services())
+ self.assertEqual([response.service_id], controller.list_ephemeral_hidden_services(detached = True))
+
+ # drop and recreate the service
+
+ self.assertEqual(True, controller.remove_ephemeral_hidden_service(response.service_id))
+ self.assertEqual([], controller.list_ephemeral_hidden_services(detached = True))
+ controller.create_ephemeral_hidden_service(4567, key_type = response.private_key_type, key_content = response.private_key, detached = True)
+ self.assertEqual([response.service_id], controller.list_ephemeral_hidden_services(detached = True))
+
+ # other controllers should be able to see this service, and drop it
+
+ with runner.get_tor_controller() as second_controller:
+ self.assertEqual([response.service_id], second_controller.list_ephemeral_hidden_services(detached = True))
+ self.assertEqual(True, second_controller.remove_ephemeral_hidden_service(response.service_id))
+ self.assertEqual([], controller.list_ephemeral_hidden_services(detached = True))
+
+ # recreate the service and confirms that it outlives this controller
+
+ response = second_controller.create_ephemeral_hidden_service(4567, detached = True)
+
+ self.assertEqual([response.service_id], controller.list_ephemeral_hidden_services(detached = True))
+ controller.remove_ephemeral_hidden_service(response.service_id)
+
+ # TODO: Uncomment the below when tor makes its 0.2.7.1 release.
+ # @require_version(Requirement.ADD_ONION)
+
+ @require_online
+ @require_controller
+ def test_using_ephemeral_hidden_services(self):
+ """
+ Create and use a live ephemeral hidden service.
+ """
+
+ # TODO: Not having success getting... well, just about any damn hidden
+ # serivce working. Even our prior tutorial is failing right now. >:(
+
+ return
+
+ with test.runner.get_runner().get_tor_controller() as controller:
+ incoming_socket, incoming_address = None, None
+
+ def run_server():
+ serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ serversocket.bind(('localhost', 4567))
+ serversocket.listen(5)
+ incoming_socket, incoming_address = serversocket.accept()
+ incoming_socket.write('hello world')
+ serversocket.shutdown(socket.SHUT_RDWR)
+
+ server_thread = threading.Thread(target = run_server)
+ server_thread.setDaemon(True)
+ server_thread.start()
+
+ response = controller.create_ephemeral_hidden_service({80: 4567})
+
+ with test.network.Socks(controller.get_socks_listeners()[0]) as s:
+ s.settimeout(30)
+ s.connect(('%s.onion' % response.service_id, 80))
+ print s.read()
+
+ server_thread.join()
+
@require_controller
def test_set_conf(self):
"""
diff --git a/test/settings.cfg b/test/settings.cfg
index c963630..fe0468c 100644
--- a/test/settings.cfg
+++ b/test/settings.cfg
@@ -178,6 +178,7 @@ test.unit_tests
|test.unit.version.TestVersion
|test.unit.tutorial.TestTutorial
|test.unit.tutorial_examples.TestTutorialExamples
+|test.unit.response.add_onion.TestAddOnionResponse
|test.unit.response.control_message.TestControlMessage
|test.unit.response.control_line.TestControlLine
|test.unit.response.events.TestEvents
diff --git a/test/unit/response/add_onion.py b/test/unit/response/add_onion.py
new file mode 100644
index 0000000..b58fe3b
--- /dev/null
+++ b/test/unit/response/add_onion.py
@@ -0,0 +1,95 @@
+"""
+Unit tests for the stem.response.add_onion.AddOnionResponse class.
+"""
+
+import unittest
+
+import stem
+import stem.response
+import stem.response.add_onion
+
+from test import mocking
+
+WITH_PRIVATE_KEY = """250-ServiceID=gfzprpioee3hoppz
+250-PrivateKey=RSA1024:MIICXgIBAAKBgQDZvYVxvKPTWhId/8Ss9fVxjAoFDsrJ3pk6HjHrEFRm3ypkK/vArbG9BrupzzYcyms+lO06O8b/iOSHuZI5mUEGkrYqQ+hpB2SkPUEzW7vcp8SQQivna3+LfkWH4JDqfiwZutU6MMEvU6g1OqK4Hll6uHbLpsfxkS/mGjyu1C9a9wIDAQABAoGBAJxsC3a25xZJqaRFfxwmIiptSTFy+/nj4T4gPQo6k/fHMKP/+P7liT9bm+uUwbITNNIjmPzxvrcKt+pNRR/92fizxr8QXr8l0ciVOLerbvdqvVUaQ/K1IVsblOLbactMvXcHactmqqLFUaZU9PPSDla7YkzikLDIUtHXQBEt4HEhAkEA/c4n+kpwi4odCaF49ESPbZC/Qejh7U9Tq10vAHzfrrGgQjnLw2UGDxJQXc9P12fGTvD2q3Q3VaMI8TKKFqZXsQJBANufh1zfP+xX/UfxJ4QzDUCHCu2gnyTDj3nG9Bc80E5g7NwR2VBXF1R+QQCK9GZcXd2y6vBYgrHOSUiLbVjGrycCQQDpOcs0zbjUEUuTsQUT+fiO50dJSrZpus6ZFxz85sMppeItWSzsVeYWbW7adYnZ2Gu72OPjM/0xPYsXEakhHSRRAkAxlVauNQjthv/72god4pi/VL224GiNmEkwKSa6iFRPHbrcBHuXk9IElWx/ft+mrHvUraw1DwaStgv9gNzzCghJAkEA08RegCRnIzuGvgeejLk4suIeCMD/11AvmSvxbRWS5rq1leSVo7uGLSnqDbwlzE4dGb5kH15NNAp14/l2Fu/yZg==
+250 OK"""
+
+WITHOUT_PRIVATE_KEY = """250-ServiceID=gfzprpioee3hoppz
+250 OK"""
+
+WRONG_FIRST_KEY = """250-MyKey=gfzprpioee3hoppz
+250-ServiceID=gfzprpioee3hoppz
+250 OK"""
+
+MISSING_KEY_TYPE = """250-ServiceID=gfzprpioee3hoppz
+250-PrivateKey=MIICXgIBAAKBgQDZvYVxvKPTWhId/8Ss9fVxj
+250 OK"""
+
+
+class TestAddOnionResponse(unittest.TestCase):
+ def test_convert(self):
+ """
+ Exercises functionality of the convert method both when it works and
+ there's an error.
+ """
+
+ # working case
+ response = mocking.get_message(WITH_PRIVATE_KEY)
+ stem.response.convert('ADD_ONION', response)
+
+ # now this should be a AddOnionResponse (ControlMessage subclass)
+ self.assertTrue(isinstance(response, stem.response.ControlMessage))
+ self.assertTrue(isinstance(response, stem.response.add_onion.AddOnionResponse))
+
+ # exercise some of the ControlMessage functionality
+ raw_content = (WITH_PRIVATE_KEY + '\n').replace('\n', '\r\n')
+ self.assertEqual(raw_content, response.raw_content())
+ self.assertTrue(str(response).startswith('ServiceID='))
+
+ def test_with_private_key(self):
+ """
+ Checks a response when there's a private key.
+ """
+
+ response = mocking.get_message(WITH_PRIVATE_KEY)
+ stem.response.convert('ADD_ONION', response)
+
+ self.assertEqual('gfzprpioee3hoppz', response.service_id)
+ self.assertTrue(response.private_key.startswith('MIICXgIBAAKB'))
+ self.assertEqual('RSA1024', response.private_key_type)
+
+ def test_without_private_key(self):
+ """
+ Checks a response without a private key.
+ """
+
+ response = mocking.get_message(WITHOUT_PRIVATE_KEY)
+ stem.response.convert('ADD_ONION', response)
+
+ self.assertEqual('gfzprpioee3hoppz', response.service_id)
+ self.assertEqual(None, response.private_key)
+ self.assertEqual(None, response.private_key_type)
+
+ def test_without_service_id(self):
+ """
+ Checks a response that lack an initial service id.
+ """
+
+ try:
+ response = mocking.get_message(WRONG_FIRST_KEY)
+ stem.response.convert('ADD_ONION', response)
+ self.fail("we should've raised a ProtocolError")
+ except stem.ProtocolError as exc:
+ self.assertTrue(str(exc).startswith('ADD_ONION response should start with'))
+
+ def test_no_key_type(self):
+ """
+ Checks a response that's missing the private key type.
+ """
+
+ try:
+ response = mocking.get_message(MISSING_KEY_TYPE)
+ stem.response.convert('ADD_ONION', response)
+ self.fail("we should've raised a ProtocolError")
+ except stem.ProtocolError as exc:
+ self.assertTrue(str(exc).startswith('ADD_ONION PrivateKey lines should be of the form'))
More information about the tor-commits
mailing list