commit a13cafd79f40433c1692ce925fc6418b33eed507 Author: Damian Johnson atagar@torproject.org Date: Sat Oct 19 13:32:03 2019 -0700
IntroductionPointV3 key methods
Simple helpers for cryptographic keys and a few tests. George's test still passes, but I haven't yet added the encoding support his class has. --- stem/descriptor/hidden_service.py | 111 ++++++++++++++++++++++-------- test/unit/descriptor/hidden_service_v3.py | 63 +++++++++++++++-- 2 files changed, 140 insertions(+), 34 deletions(-)
diff --git a/stem/descriptor/hidden_service.py b/stem/descriptor/hidden_service.py index cfa20d09..c2673fb3 100644 --- a/stem/descriptor/hidden_service.py +++ b/stem/descriptor/hidden_service.py @@ -73,6 +73,13 @@ if stem.prereq._is_lru_cache_available(): else: from stem.util.lru_cache import lru_cache
+try: + from cryptography.hazmat.backends.openssl.backend import backend + X25519_AVAILABLE = backend.x25519_supported() +except ImportError: + X25519_AVAILABLE = False + + REQUIRED_V2_FIELDS = ( 'rendezvous-service-descriptor', 'version', @@ -144,21 +151,76 @@ class IntroductionPoints(collections.namedtuple('IntroductionPoints', INTRODUCTI """
-class IntroductionPointV3(collections.namedtuple('IntroductionPointV3', ['link_specifiers', 'onion_key', 'auth_key', 'enc_key', 'enc_key_cert', 'legacy_key', 'legacy_key_cert'])): +class IntroductionPointV3(collections.namedtuple('IntroductionPointV3', ['link_specifiers', 'onion_key_raw', 'auth_key_cert', 'enc_key_raw', 'enc_key_cert', 'legacy_key_raw', 'legacy_key_cert'])): """ Introduction point for a v3 hidden service.
.. versionadded:: 1.8.0
:var list link_specifiers: :class:`~stem.client.datatype.LinkSpecifier` where this service is reachable - :var str onion_key: ntor introduction point public key - :var str auth_key: cross-certifier of the signing key - :var str enc_key: introduction request encryption key - :var str enc_key_cert: cross-certifier of the signing key by the encryption key - :var str legacy_key: legacy introduction point RSA public key - :var str legacy_key_cert: cross-certifier of the signing key by the legacy key + :var str onion_key_raw: base64 ntor introduction point public key + :var stem.certificate.Ed25519Certificate auth_key_cert: cross-certifier of the signing key with the auth key + :var str enc_key_raw: base64 introduction request encryption key + :var stem.certificate.Ed25519Certificate enc_key_cert: cross-certifier of the signing key by the encryption key + :var str legacy_key_raw: base64 legacy introduction point RSA public key + :var stem.certificate.Ed25519Certificate legacy_key_cert: cross-certifier of the signing key by the legacy key """
+ def onion_key(self): + """ + Provides our ntor introduction point public key. + + :returns: ntor :class:`~cryptography.hazmat.primitives.asymmetric.x25519.X25519PublicKey` + + :raises: + * **ImportError** if required the cryptography module is unavailable + * **EnvironmentError** if OpenSSL x25519 unsupported + """ + + return IntroductionPointV3._parse_key(self.onion_key_raw) + + def enc_key(self): + """ + Provides our encryption key. + + :returns: encryption :class:`~cryptography.hazmat.primitives.asymmetric.x25519.X25519PublicKey` + + :raises: + * **ImportError** if required the cryptography module is unavailable + * **EnvironmentError** if OpenSSL x25519 unsupported + """ + + return IntroductionPointV3._parse_key(self.enc_key_raw) + + def legacy_key(self): + """ + Provides our legacy introduction point public key. + + :returns: legacy :class:`~cryptography.hazmat.primitives.asymmetric.x25519.X25519PublicKey` + + :raises: + * **ImportError** if required the cryptography module is unavailable + * **EnvironmentError** if OpenSSL x25519 unsupported + """ + + return IntroductionPointV3._parse_key(self.legacy_key_raw) + + @staticmethod + def _parse_key(value): + if value is None: + return value + elif not stem.prereq.is_crypto_available(): + raise ImportError('cryptography module unavailable') + elif not X25519_AVAILABLE: + # without this the cryptography raises... + # cryptography.exceptions.UnsupportedAlgorithm: X25519 is not supported by this version of OpenSSL. + + raise EnvironmentError('OpenSSL x25519 unsupported') + + from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PublicKey + + return X25519PublicKey.from_public_bytes(base64.b64decode(value)) +
class AlternateIntroductionPointV3(object): """ @@ -425,9 +487,6 @@ def _parse_v3_inner_formats(descriptor, entries):
def _parse_v3_introduction_points(descriptor, entries): - from cryptography.hazmat.backends.openssl.backend import backend - from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PublicKey - if hasattr(descriptor, '_unparsed_introduction_points'): introduction_points = [] remaining = descriptor._unparsed_introduction_points @@ -445,12 +504,8 @@ def _parse_v3_introduction_points(descriptor, entries): entry = _descriptor_components(intro_point_str, False) link_specifiers = _parse_link_specifiers(_value('introduction-point', entry))
- if backend.x25519_supported(): - onion_key_line = _value('onion-key', entry) - onion_key_b64 = onion_key_line[5:] if onion_key_line.startswith('ntor ') else None - onion_key = X25519PublicKey.from_public_bytes(base64.b64decode(onion_key_b64)) - else: - onion_key = None + onion_key_line = _value('onion-key', entry) + onion_key = onion_key_line[5:] if onion_key_line.startswith('ntor ') else None
_, block_type, auth_key_cert = entry['auth-key'][0] auth_key_cert = Ed25519Certificate.parse(auth_key_cert) @@ -458,12 +513,8 @@ def _parse_v3_introduction_points(descriptor, entries): if block_type != 'ED25519 CERT': raise ValueError('Expected auth-key to have an ed25519 certificate, but was %s' % block_type)
- if backend.x25519_supported(): - enc_key_line = _value('enc-key', entry) - enc_key_b64 = enc_key_line[5:] if enc_key_line.startswith('ntor ') else None - enc_key = X25519PublicKey.from_public_bytes(base64.b64decode(enc_key_b64)) - else: - enc_key = None + enc_key_line = _value('enc-key', entry) + enc_key = enc_key_line[5:] if enc_key_line.startswith('ntor ') else None
_, block_type, enc_key_cert = entry['enc-key-cert'][0] enc_key_cert = Ed25519Certificate.parse(enc_key_cert) @@ -475,14 +526,14 @@ def _parse_v3_introduction_points(descriptor, entries): legacy_key_cert = entry['legacy-key-cert'][0][2] if 'legacy-key-cert' in entry else None
introduction_points.append( - AlternateIntroductionPointV3( - link_specifiers = link_specifiers, - onion_key = onion_key, - auth_key_cert = auth_key_cert, - enc_key = enc_key, - enc_key_cert = enc_key_cert, - legacy_key = legacy_key, - legacy_key_cert = legacy_key_cert, + IntroductionPointV3( + link_specifiers, + onion_key, + auth_key_cert, + enc_key, + enc_key_cert, + legacy_key, + legacy_key_cert, ) )
diff --git a/test/unit/descriptor/hidden_service_v3.py b/test/unit/descriptor/hidden_service_v3.py index 7e87668d..499064f9 100644 --- a/test/unit/descriptor/hidden_service_v3.py +++ b/test/unit/descriptor/hidden_service_v3.py @@ -14,6 +14,7 @@ import stem.prereq from stem.descriptor.hidden_service import ( CHECKSUM_CONSTANT, REQUIRED_V3_FIELDS, + X25519_AVAILABLE, AlternateIntroductionPointV3, HiddenServiceDescriptorV3, OuterLayer, @@ -26,6 +27,12 @@ from test.unit.descriptor import ( base_expect_invalid_attr_for_text, )
+try: + # added in python 3.3 + from unittest.mock import patch, Mock, MagicMock +except ImportError: + from mock import patch, Mock, MagicMock + expect_invalid_attr = functools.partial(base_expect_invalid_attr, HiddenServiceDescriptorV3, 'version', 3) expect_invalid_attr_for_text = functools.partial(base_expect_invalid_attr_for_text, HiddenServiceDescriptorV3, 'version', 3)
@@ -264,6 +271,51 @@ class TestHiddenServiceDescriptorV3(unittest.TestCase): self.assertRaisesWith(ValueError, "'boom.onion' isn't a valid hidden service v3 address", HiddenServiceDescriptorV3._public_key_from_address, 'boom') self.assertRaisesWith(ValueError, 'Bad checksum (expected def7 but was 842e)', HiddenServiceDescriptorV3._public_key_from_address, '5' * 56)
+ def test_intro_point_crypto(self): + """ + Retrieve IntroductionPointV3 cryptographic materials. + """ + + if not stem.prereq.is_crypto_available(): + self.skipTest('(requires cryptography support)') + return + elif not X25519_AVAILABLE: + self.skipTest('(openssl requires X25519 support)') + return + + from cryptography.hazmat.backends.openssl.x25519 import X25519PublicKey + from cryptography.hazmat.primitives import serialization + + intro_point = InnerLayer(INNER_LAYER_STR).introduction_points[0] + + self.assertEqual('AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=', intro_point.onion_key_raw) + self.assertEqual('AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=', intro_point.enc_key_raw) + + self.assertTrue(isinstance(intro_point.onion_key(), X25519PublicKey)) + self.assertTrue(isinstance(intro_point.enc_key(), X25519PublicKey)) + + self.assertEqual(intro_point.onion_key_raw, base64.b64encode(intro_point.onion_key().public_bytes( + encoding = serialization.Encoding.Raw, + format = serialization.PublicFormat.Raw, + ))) + + self.assertEqual(intro_point.enc_key_raw, base64.b64encode(intro_point.enc_key().public_bytes( + encoding = serialization.Encoding.Raw, + format = serialization.PublicFormat.Raw, + ))) + + self.assertEqual(None, intro_point.legacy_key_raw) + self.assertEqual(None, intro_point.legacy_key()) + + @patch('stem.prereq.is_crypto_available', Mock(return_value = False)) + def test_intro_point_crypto_without_prereq(self): + """ + Fetch cryptographic materials when the module is unavailable. + """ + + intro_point = InnerLayer(INNER_LAYER_STR).introduction_points[0] + self.assertRaisesWith(ImportError, 'cryptography module unavailable', intro_point.onion_key) + def test_encode_decode_descriptor(self): """ Encode an HSv3 descriptor and then decode it and make sure you get the intended results. @@ -277,7 +329,7 @@ class TestHiddenServiceDescriptorV3(unittest.TestCase): self.skipTest('(requires cryptography ed25519 support)') return
- from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey, Ed25519PrivateKey from cryptography.hazmat.primitives import serialization
# Build the service @@ -317,9 +369,12 @@ class TestHiddenServiceDescriptorV3(unittest.TestCase): for original_intro in intro_points: # Match intro points
- if _pubkeys_are_equal(desc_intro.auth_key, original_intro.auth_key): + auth_key_1 = Ed25519PublicKey.from_public_bytes(desc_intro.auth_key_cert.key) + auth_key_2 = original_intro.auth_key + + if _pubkeys_are_equal(auth_key_1, auth_key_2): original_found = True - self.assertTrue(_pubkeys_are_equal(desc_intro.enc_key, original_intro.enc_key)) - self.assertTrue(_pubkeys_are_equal(desc_intro.onion_key, original_intro.onion_key)) + self.assertTrue(_pubkeys_are_equal(desc_intro.enc_key(), original_intro.enc_key)) + self.assertTrue(_pubkeys_are_equal(desc_intro.onion_key(), original_intro.onion_key))
self.assertTrue(original_found)