[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]
[tor-commits] [stem/master] Initial KeyCertificate implementation
commit 6ee5abbe5c1137b4a311d5b7ef5003874982008e
Author: Damian Johnson <atagar@xxxxxxxxxxxxxx>
Date: Sun Sep 23 14:20:49 2012 -0700
Initial KeyCertificate implementation
First stab at parsing the authority section's key certificates. This is
completely untested, next step is to write some unit tests for it.
---
stem/descriptor/networkstatus.py | 170 ++++++++++++++++++++++++++++++++++++++
1 files changed, 170 insertions(+), 0 deletions(-)
diff --git a/stem/descriptor/networkstatus.py b/stem/descriptor/networkstatus.py
index 17a9877..46adbfe 100644
--- a/stem/descriptor/networkstatus.py
+++ b/stem/descriptor/networkstatus.py
@@ -106,6 +106,21 @@ DEFAULT_PARAMS = {
"cbtinitialtimeout": 60000,
}
+# KeyCertificate fields, tuple is of the form...
+# (keyword, is_mandatory)
+
+KEY_CERTIFICATE_PARAMS = (
+ ('dir-key-certificate-version', True),
+ ('dir-address', False),
+ ('fingerprint', True),
+ ('dir-identity-key', True),
+ ('dir-key-published', True),
+ ('dir-key-expires', True),
+ ('dir-signing-key', True),
+ ('dir-key-crosscert', False),
+ ('dir-key-certification', True),
+)
+
BANDWIDTH_WEIGHT_ENTRIES = (
"Wbd", "Wbe", "Wbg", "Wbm",
"Wdb",
@@ -768,6 +783,161 @@ class DirectoryAuthority(stem.descriptor.Descriptor):
return self.unrecognized_lines
+class KeyCertificate(stem.descriptor.Descriptor):
+ """
+ Directory key certificate for a v3 network status document.
+
+ :var int version: **\*** version of the key certificate
+ :var str address: authority's IP address
+ :var int dir_port: authority's DirPort
+ :var str fingerprint: **\*** authority's fingerprint
+ :var str identity_key: **\*** long term authority identity key
+ :var datetime published: **\*** time when this key was generated
+ :var datetime expires: **\*** time after which this key becomes invalid
+ :var str signing_key: **\*** directory server's public signing key
+ :var str crosscert: signature made using certificate's signing key
+ :var str certification: **\*** signature of this key certificate signed with the identity key
+
+ **\*** mandatory attribute
+ """
+
+ def __init__(self, raw_content, validate):
+ super(KeyCertificate, self).__init__(raw_content)
+
+ self.version = None
+ self.address = None
+ self.dir_port = None
+ self.fingerprint = None
+ self.identity_key = None
+ self.published = None
+ self.expires = None
+ self.signing_key = None
+ self.crosscert = None
+ self.certification = None
+
+ self._unrecognized_lines = []
+
+ self._parse(raw_contents, validate)
+
+ def _parse(self, content, validate):
+ """
+ Parses the given content and applies the attributes.
+
+ :param str content: descriptor content
+ :param bool validate: checks validity if True
+
+ :raises: ValueError if a validity check fails
+ """
+
+ entries, first_keyword, last_keyword, _ = stem.descriptor._get_descriptor_components(content, validate)
+
+ if validate:
+ if first_keyword != 'dir-key-certificate-version':
+ raise ValueError("Key certificates must start with a 'dir-key-certificate-version' line:\n%s" % (content))
+ elif last_keyword != 'dir-key-certification':
+ raise ValueError("Key certificates must end with a 'dir-key-certification' line:\n%s" % (content))
+
+ # check that we have mandatory fields and that our known fields only
+ # appear once
+
+ for keyword, is_mandatory in KEY_CERTIFICATE_PARAMS:
+ if is_mandatory and not keyword in entries:
+ raise ValueError("Key certificates must have a '%s' line:\n%s" % (keyword, content))
+
+ entry_count = len(entries.get(keyword, []))
+ if entry_count > 1:
+ raise ValueError("Key certificates can only have a single '%s' line, got %i:\n%s" % (keyword, entry_count, content))
+
+ # Check that our field's order matches the spec. This isn't explicitely
+ # stated in the spec, but the network status document requires a specific
+ # order so it stands to reason that the key certificate (which is in it)
+ # needs ot match a prescribed order too.
+
+ fields = [attr[0] for attr in KEY_CERTIFICATE_PARAMS]
+ _check_for_misordered_fields(entries, fields)
+
+ for keyword, values in entries.items():
+ value, block_contents = values[0]
+ line = "%s %s" % (keyword, value)
+
+ if keyword == 'dir-key-certificate-version':
+ # "dir-key-certificate-version" version
+
+ if not value.isdigit():
+ if not validate: continue
+ raise ValueError("Key certificate has a non-integer version: %s" % line)
+
+ self.version = int(value)
+
+ if validate and self.version != 3:
+ raise ValueError("Expected a version 3 key certificate, got version '%i' instead" % self.version)
+ elif keyword == 'dir-address':
+ # "dir-address" IPPort
+
+ if not ':' in value:
+ if not validate: continue
+ raise ValueError("Key certificate's 'dir-address' is expected to be of the form ADDRESS:PORT: %s" % line)
+
+ address, dirport = value.split(':', 1)
+
+ if validate:
+ if not stem.util.connection.is_valid_ip_address(address):
+ raise ValueError("Key certificate's address isn't a valid IPv4 address: %s" % line)
+ elif not stem.util.connection.is_valid_port(dirport):
+ raise ValueError("Key certificate's dirport is invalid: %s" % line)
+
+ self.address = address
+ self.dir_port = dirport
+ elif keyword == 'fingerprint':
+ # "fingerprint" fingerprint
+
+ if validate and not stem.util.tor_tools.is_valid_fingerprint(value):
+ raise ValueError("Key certificate's fingerprint is malformed: %s" % line)
+
+ self.fingerprint = value
+ elif keyword in ('dir-key-published', 'dir-key-expires'):
+ # "dir-key-published" YYYY-MM-DD HH:MM:SS
+ # "dir-key-expires" YYYY-MM-DD HH:MM:SS
+
+ try:
+ date_value = datetime.datetime.strptime(value, "%Y-%m-%d %H:%M:%S")
+
+ if keyword == 'dir-key-published':
+ self.published = date_value
+ elif keyword == 'dir-key-expires':
+ self.expires = date_value
+ except ValueError:
+ if validate:
+ raise ValueError("Key certificate's '%s' time wasn't parseable: %s" % (keyword, value))
+ elif keyword in ('dir-identity-key', 'dir-signing-key', 'dir-key-crosscert', 'dir-key-certification'):
+ # "dir-identity-key" NL a public key in PEM format
+ # "dir-signing-key" NL a key in PEM format
+ # "dir-key-crosscert" NL CrossSignature
+ # "dir-key-certification" NL Signature
+
+ if validate and not block_contents:
+ raise ValueError("Key certificate's '%s' line must be followed by a key block: %s" % (keyword, line))
+
+ if keyword == 'dir-identity-key':
+ self.identity_key = block_contents
+ elif keyword == 'dir-signing-key':
+ self.signing_key = block_contents
+ elif keyword == 'dir-key-crosscert':
+ self.crosscert = block_contents
+ elif keyword == 'dir-key-certification':
+ self.certification = block_contents
+ else:
+ self._unrecognized_lines.append(line)
+
+ def get_unrecognized_lines(self):
+ """
+ Returns any unrecognized lines.
+
+ :returns: a list of unrecognized lines
+ """
+
+ return self.unrecognized_lines
+
# TODO: microdescriptors have a slightly different format (including a
# 'method') - should probably be a subclass
class DocumentSignature(object):
_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits