[tor-commits] [stem/master] Drop sign_descriptor_content() testing helper
atagar at torproject.org
atagar at torproject.org
Tue May 2 19:57:29 UTC 2017
commit 6bc661515d20e733b730ea6722a4e3d3e72adddb
Author: Damian Johnson <atagar at torproject.org>
Date: Sun Apr 30 23:36:42 2017 -0700
Drop sign_descriptor_content() testing helper
This helper wasn't used by our tests any longer, and on reflection I can't make
the router-signature it generates pass validation. There's one clear bug that
slipped in a while ago: "b'b'.join()" should be "b'\n'.join()". But correcting
that didn't make the function work.
For now just dropping the broken helper. It would be sweeeet to be able to sign
the descriptors we create but that's gonna take a bit of work.
---
stem/descriptor/__init__.py | 13 ++---
stem/descriptor/server_descriptor.py | 6 +-
test/mocking.py | 108 -----------------------------------
test/unit/tutorial.py | 2 -
4 files changed, 7 insertions(+), 122 deletions(-)
diff --git a/stem/descriptor/__init__.py b/stem/descriptor/__init__.py
index 32b7d7c..c790e65 100644
--- a/stem/descriptor/__init__.py
+++ b/stem/descriptor/__init__.py
@@ -530,7 +530,7 @@ class Descriptor(object):
self._unrecognized_lines = []
@classmethod
- def content(cls, attr = None, exclude = (), sign = False):
+ def content(cls, attr = None, exclude = ()):
"""
Creates descriptor content with the given attributes. Mandatory fields are
filled with dummy information unless data is supplied.
@@ -540,19 +540,16 @@ class Descriptor(object):
:param dict attr: keyword/value mappings to be included in the descriptor
:param list exclude: mandatory keywords to exclude from the descriptor, this
results in an invalid descriptor
- :param bool sign_content: includes cryptographic digest if True
:returns: **str** with the content of a descriptor
- :raises:
- * **ImportError** if cryptography is unavailable and sign is True
- * **NotImplementedError** if not implemented for this descriptor type
+ :raises: **NotImplementedError** if not implemented for this descriptor type
"""
raise NotImplementedError("The create and content methods haven't been implemented for %s" % cls.__name__)
@classmethod
- def create(cls, attr = None, exclude = (), validate = True, sign = False):
+ def create(cls, attr = None, exclude = (), validate = True):
"""
Creates a descriptor with the given attributes. Mandatory fields are filled
with dummy information unless data is supplied.
@@ -564,17 +561,15 @@ class Descriptor(object):
results in an invalid descriptor
:param bool validate: checks the validity of the descriptor's content if
**True**, skips these checks otherwise
- :param bool sign_content: includes cryptographic digest if True
:returns: :class:`~stem.descriptor.Descriptor` subclass
:raises:
* **ValueError** if the contents is malformed and validate is True
- * **ImportError** if cryptography is unavailable and sign is True
* **NotImplementedError** if not implemented for this descriptor type
"""
- return cls(cls.content(attr, exclude, sign), validate = validate)
+ return cls(cls.content(attr, exclude), validate = validate)
def get_path(self):
"""
diff --git a/stem/descriptor/server_descriptor.py b/stem/descriptor/server_descriptor.py
index 9f7c846..b4dcd23 100644
--- a/stem/descriptor/server_descriptor.py
+++ b/stem/descriptor/server_descriptor.py
@@ -809,12 +809,12 @@ class RelayDescriptor(ServerDescriptor):
self.certificate.validate(self)
@classmethod
- def content(cls, attr = None, exclude = (), sign = False):
+ def content(cls, attr = None, exclude = ()):
return _descriptor_content(attr, exclude, RELAY_SERVER_HEADER, RELAY_SERVER_FOOTER)
@classmethod
- def create(cls, attr = None, exclude = (), validate = True, sign = False):
- return cls(cls.content(attr, exclude, sign), validate = validate, skip_crypto_validation = not sign)
+ def create(cls, attr = None, exclude = (), validate = True):
+ return cls(cls.content(attr, exclude), validate = validate, skip_crypto_validation = True)
@lru_cache()
def digest(self):
diff --git a/test/mocking.py b/test/mocking.py
index 4843a97..990aa89 100644
--- a/test/mocking.py
+++ b/test/mocking.py
@@ -663,111 +663,3 @@ def get_network_status_document_v3(attr = None, exclude = (), authorities = None
return desc_content
else:
return stem.descriptor.networkstatus.NetworkStatusDocumentV3(desc_content, validate = True)
-
-
-def sign_descriptor_content(desc_content):
- """
- Add a valid signature to the supplied descriptor string.
-
- If cryptography is available the function will generate a key pair, and use
- it to sign the descriptor string. Any existing fingerprint, signing-key or
- router-signature data will be overwritten. If the library's unavailable the
- code will return the unaltered descriptor.
-
- :param str desc_content: the descriptor string to sign
- :returns: a descriptor string, signed if crypto available and unaltered otherwise
- """
-
- if not stem.prereq.is_crypto_available():
- return desc_content
- else:
- from cryptography.hazmat.backends import default_backend
- from cryptography.hazmat.primitives import hashes
- from cryptography.hazmat.primitives.asymmetric import rsa, padding
- from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
-
- private_key = rsa.generate_private_key(public_exponent=65537, key_size=1024, backend=default_backend())
- public_key = private_key.public_key()
-
- # Get a string representation of the public key
-
- signing_key_token = b'\nsigning-key\n' # note the trailing '\n' is important here so as not to match the string elsewhere
- signing_key_token_end = b'\n-----END RSA PUBLIC KEY-----\n'
- new_sk = signing_key_token + public_key.public_bytes(encoding=Encoding.PEM, format=PublicFormat.PKCS1)
-
- # update the descriptor string with the new signing key
-
- skt_start = desc_content.find(signing_key_token)
- skt_end = desc_content.find(signing_key_token_end, skt_start)
- desc_content = desc_content[:skt_start] + new_sk + desc_content[skt_end + len(signing_key_token_end):]
-
- # generate the new fingerprint string
-
- seq_as_string = public_key.public_bytes(encoding=Encoding.DER, format=PublicFormat.PKCS1)
- key_hash = stem.util.str_tools._to_bytes(hashlib.sha1(seq_as_string).hexdigest().upper())
- grouped_fingerprint = b''
-
- for x in range(0, len(key_hash), 4):
- grouped_fingerprint += b' ' + key_hash[x:x + 4]
- fingerprint_token = b'\nfingerprint'
- new_fp = fingerprint_token + grouped_fingerprint
-
- # update the descriptor string with the new fingerprint
-
- ft_start = desc_content.find(fingerprint_token)
- if ft_start < 0:
- fingerprint_token = b'\nopt fingerprint'
- ft_start = desc_content.find(fingerprint_token)
-
- # if the descriptor does not already contain a fingerprint do not add one
-
- if ft_start >= 0:
- ft_end = desc_content.find(b'\n', ft_start + 1)
- desc_content = desc_content[:ft_start] + new_fp + desc_content[ft_end:]
-
- # create a temporary object to use to calculate the digest
-
- tempDesc = stem.descriptor.server_descriptor.RelayDescriptor(desc_content, validate=False)
-
- # calculate the new digest for the descriptor
-
- new_digest_hex = tempDesc.digest().lower()
-
- # remove the hex encoding
-
- if stem.prereq.is_python_3():
- new_digest = bytes.fromhex(new_digest_hex)
- else:
- new_digest = new_digest_hex.decode('hex_codec')
-
- # Generate the digest buffer.
- # block is 128 bytes in size
- # 2 bytes for the type info
- # 1 byte for the separator
-
- digest_padding = b''
-
- for x in range(125 - len(new_digest)):
- digest_padding += b'\xFF'
- digest_buffer = b'\x00\x01' + digest_padding + b'\x00' + new_digest
-
- # generate a new signature by signing the digest buffer with the private key
-
- signature_as_bytes = private_key.sign(digest_buffer, padding.PKCS1v15(), hashes.SHA1())
- signature_base64 = base64.b64encode(signature_as_bytes)
-
- signature_base64 = b'b'.join([
- signature_base64[:64],
- signature_base64[64:128],
- signature_base64[128:],
- ])
-
- # update the descriptor string with the new signature
-
- router_signature_token = b'\nrouter-signature\n'
- router_signature_start = b'-----BEGIN SIGNATURE-----\n'
- router_signature_end = b'\n-----END SIGNATURE-----\n'
- rst_start = desc_content.find(router_signature_token)
- desc_content = desc_content[:rst_start] + router_signature_token + router_signature_start + signature_base64 + router_signature_end
-
- return desc_content
diff --git a/test/unit/tutorial.py b/test/unit/tutorial.py
index 26ea2b1..a404b42 100644
--- a/test/unit/tutorial.py
+++ b/test/unit/tutorial.py
@@ -207,8 +207,6 @@ class TestTutorial(unittest.TestCase):
return
exit_descriptor = RelayDescriptor.content({'router': 'speedyexit 149.255.97.109 9001 0 0'}).replace(b'reject *:*', b'accept *:*')
-
- exit_descriptor = mocking.sign_descriptor_content(exit_descriptor)
exit_descriptor = RelayDescriptor(exit_descriptor)
downloader_mock().get_server_descriptors().run.return_value = [
More information about the tor-commits
mailing list