Index: twisted/internet/interfaces.py =================================================================== --- twisted/internet/interfaces.py (revision 40421) +++ twisted/internet/interfaces.py (working copy) @@ -570,6 +570,28 @@ """ + def lookupCertificate(name, timeout=None): + """ + Perform a CERT record lookup. + + @type name: C{str} + @param name: DNS name to resolve. + + @type timeout: Sequence of C{int} + @param timeout: Number of seconds after which to reissue the query. + When the last timeout expires, the query is considered failed. + + @rtype: L{Deferred} + @return: A L{Deferred} which fires with a three-tuple of lists of + L{twisted.names.dns.RRHeader} instances. The first element of the + tuple gives answers. The second element of the tuple gives + authorities. The third element of the tuple gives additional + information. The L{Deferred} may instead fail with one of the + exceptions defined in L{twisted.names.error} or with + C{NotImplementedError}. + """ + + def lookupNamingAuthorityPointer(name, timeout=None): """ Perform a NAPTR record lookup. Index: twisted/names/client.py =================================================================== --- twisted/names/client.py (revision 40421) +++ twisted/names/client.py (working copy) @@ -725,6 +725,11 @@ +def lookupCertificate(name, timeout=None): + return getResolver().lookupCertificate(name, timeout) + + + def lookupResponsibility(name, timeout=None): return getResolver().lookupResponsibility(name, timeout) Index: twisted/names/common.py =================================================================== --- twisted/names/common.py (revision 40421) +++ twisted/names/common.py (working copy) @@ -149,6 +149,10 @@ return self._lookup(name, dns.IN, dns.SPF, timeout) + def lookupCertificate(self, name, timeout=None): + return self._lookup(name, dns.IN, dns.CERT, timeout) + + def lookupResponsibility(self, name, timeout=None): return self._lookup(name, dns.IN, dns.RP, timeout) @@ -240,6 +244,7 @@ dns.MX: 'lookupMailExchange', dns.TXT: 'lookupText', dns.SPF: 'lookupSenderPolicy', + dns.CERT: 'lookupCertificate', dns.RP: 'lookupResponsibility', dns.AFSDB: 'lookupAFSDatabase', Index: twisted/names/dns.py =================================================================== --- twisted/names/dns.py (revision 40421) +++ twisted/names/dns.py (working copy) @@ -14,7 +14,7 @@ __all__ = [ 'IEncodable', 'IRecord', - 'A', 'A6', 'AAAA', 'AFSDB', 'CNAME', 'DNAME', 'HINFO', + 'A', 'A6', 'AAAA', 'AFSDB', 'CERT', 'CNAME', 'DNAME', 'HINFO', 'MAILA', 'MAILB', 'MB', 'MD', 'MF', 'MG', 'MINFO', 'MR', 'MX', 'NAPTR', 'NS', 'NULL', 'OPT', 'PTR', 'RP', 'SOA', 'SPF', 'SRV', 'TXT', 'WKS', @@ -44,7 +44,7 @@ # System imports -import struct, random, socket +import base64, struct, random, socket from itertools import chain from io import BytesIO @@ -116,6 +116,7 @@ AAAA = 28 SRV = 33 NAPTR = 35 +CERT = 37 A6 = 38 DNAME = 39 OPT = 41 @@ -146,6 +147,7 @@ AAAA: 'AAAA', SRV: 'SRV', NAPTR: 'NAPTR', + CERT: 'CERT', A6: 'A6', DNAME: 'DNAME', OPT: 'OPT', @@ -1555,7 +1557,120 @@ +def _base64Format(bytes): + """ + Base64 encode C{bytes} without any line breaks + + @param bytes: The bytestring to encode. + @type bytes: L{bytes} + + @return: The formatted base64 bytestring. + """ + return nativeString(base64.encodestring(bytes).replace(b'\n', b'')) + + + @implementer(IEncodable, IRecord) +class Record_CERT(tputil.FancyStrMixin, tputil.FancyEqMixin): + """ + A Certificate record. + + @see: U{http://tools.ietf.org/html/rfc4398} + + @ivar TYPE: CERT type code constant C{37}. + @ivar fancybasename: See L{tputil.FancyStrMixin} + @ivar showAttributes: See L{tputil.FancyStrMixin} + @ivar compareAttributes: See L{tputil.FancyEqMixin} + + @ivar certType: See L{__init__} + @ivar keyTag: See L{__init__} + @ivar algorithm: See L{__init__} + @ivar certOrCRL: See L{__init__} + @ivar ttl: See L{__init__} + """ + + TYPE = CERT + + fancybasename = 'CERT' + + showAttributes = ( + 'certType', 'keyTag', 'algorithm', + ('certOrCRL', _base64Format), 'ttl') + + compareAttributes = ( + 'certType', 'keyTag', 'algorithm', + 'certOrCRL', 'ttl') + + _fmt = '!HHB' + _fmt_size = struct.calcsize(_fmt) + + def __init__(self, certType=1, keyTag=0, + algorithm=5, certOrCRL=b'', ttl=None): + """ + @param certType: an L{int} representing the certificate type + used in this CERT record. The default value (C{1}) represents + X.509 as per PKIX. See + U{https://www.iana.org/assignments/cert-rr-types/cert-rr-types.xhtml} + @type certType: L{int} + + @param keyTag: Key tag value as described in RRSIG. + The default value (C{0}) is recommende if the + algorithm field is 0. See + U{https://tools.ietf.org/html/rfc4034#appendix-B} + @type keyTag: L{int} + + @param algorithm: an L{int} representing the algorithm number + used in this CERT. The default value (C{5}) represents + RSA/SHA-1. These values are the same as used for DNSSEC. See + U{https://www.iana.org/assignments/dns-sec-alg-numbers/dns-sec-alg-numbers.txt} + @type algorithm: L{int} + + @param certOrCRL: a base64 encoded string representing the + stored certificate or CRL. + @type certOrCRL: L{bytes} + + @param ttl: The time-to-live of this record. TTL can be + supplied as an L{int} representing a period in seconds or + a human readable string can be supplied which will be + parsed by L{str2time}. Default is C{None}. + @type ttl: L{int} or L{str} or C{None}. + """ + self.certType = certType + self.keyTag = keyTag + self.algorithm = algorithm + self.certOrCRL = certOrCRL + self.ttl = str2time(ttl) + + + def encode(self, strio, compDict=None): + strio.write( + struct.pack(self._fmt, self.certType, self.keyTag, self.algorithm)) + strio.write(self.certOrCRL) + + + def decode(self, strio, length=None): + hdr = readPrecisely(strio, self._fmt_size) + self.certType, self.keyTag, self.algorithm = struct.unpack(self._fmt, hdr) + + length -= self._fmt_size + self.certOrCRL = readPrecisely(strio, length) + + + def __hash__(self): + """ + A has allowing this L{Record_CERT} to be used as a L{dict} + key. + + @return: A L{hash} og the values of + L{Record_CERT.compareAttributes} except C{ttl}. + """ + return hash(tuple(getattr(self, k) + for k in self.compareAttributes + if k != 'ttl')) + + + +@implementer(IEncodable, IRecord) class Record_AFSDB(tputil.FancyStrMixin, tputil.FancyEqMixin): """ Map from a domain name to the name of an AFS cell database server. Index: twisted/names/test/test_client.py =================================================================== --- twisted/names/test/test_client.py (revision 40421) +++ twisted/names/test/test_client.py (working copy) @@ -974,6 +974,14 @@ d.addCallback(self.checkResult, dns.SPF) return d + def test_lookupCertificate(self): + """ + See L{test_lookupAddress} + """ + d = client.lookupCertificate(self.hostname) + d.addCallback(self.checkResult, dns.CERT) + return d + def test_lookupResponsibility(self): """ See L{test_lookupAddress} Index: twisted/names/test/test_dns.py =================================================================== --- twisted/names/test/test_dns.py (revision 40421) +++ twisted/names/test/test_dns.py (working copy) @@ -30,6 +30,7 @@ dns.Record_WKS, dns.Record_SRV, dns.Record_AFSDB, dns.Record_RP, dns.Record_HINFO, dns.Record_MINFO, dns.Record_MX, dns.Record_TXT, dns.Record_AAAA, dns.Record_A6, dns.Record_NAPTR, dns.UnknownRecord, + dns.Record_CERT, ] @@ -570,7 +571,15 @@ self._recordRoundtripTest(dns.Record_TXT(b'foo', b'bar')) + def test_CERT(self): + """ + The byte stream written by L{dns.Record_CERT.encode} can be used by + L{dns.Record_CERT.decode} to reconstruct the state of the original + L{dns.Record_Cert} instance. + """ + self._recordRoundtripTest(dns.Record_CERT(certOrCRL=b'foobar')) + MESSAGE_AUTHENTIC_DATA_BYTES = ( b'\x00\x00' # ID b'\x00' # @@ -1291,6 +1300,16 @@ "") + def test_cert(self): + """ + The repr of a L{dns.Record_CERT} instance includes the data and ttl + fields of the record. + """ + self.assertEqual( + repr(dns.Record_CERT(certOrCRL=b'foobar', ttl=15)), + "") + + def test_unknown(self): """ The repr of a L{dns.UnknownRecord} instance includes the data and ttl @@ -1850,6 +1869,39 @@ dns.Record_SPF('foo', 'bar', ttl=100)) + def test_cert(self): + """ + L{dns.Record_CERT} instances compare equal if and only if + they have the same certType, keyTag, algorithm, + certOrCRL, and ttl. + """ + self._equalityTest( + dns.Record_CERT(certType=1), + dns.Record_CERT(certType=1), + dns.Record_CERT(certType=2)) + + self._equalityTest( + dns.Record_CERT(keyTag=1), + dns.Record_CERT(keyTag=1), + dns.Record_CERT(keyTag=2)) + + self._equalityTest( + dns.Record_CERT(algorithm=1), + dns.Record_CERT(algorithm=1), + dns.Record_CERT(algorithm=2)) + + self._equalityTest( + dns.Record_CERT(certOrCRL=b'foo'), + dns.Record_CERT(certOrCRL=b'foo'), + dns.Record_CERT(certOrCRL=b'bar')) + + self._equalityTest( + dns.Record_CERT(ttl=10), + dns.Record_CERT(ttl=10), + dns.Record_CERT(ttl=100)) + + + def test_unknown(self): """ L{dns.UnknownRecord} instances compare equal if and only if they have @@ -2619,3 +2671,179 @@ o.decode(b) self.assertEqual(o.code, 1) self.assertEqual(o.data, b'foobar') + + +class CERT_TEST_DATA(object): + """ + Generate byte and instance representations of an + L{dns.Record_CERT} where all attributes are set to non-default + values. + + For testing whether attributes have really been read from the byte + string during decoding. + """ + @classmethod + def BYTES(cls): + """ + @return: L{bytes} representing the encoded CERT record returned + by L{OBJECT}. + """ + return ( + b'\x00\x02' # certType + b'\x00\x01' # keyTag + b'\x04' # algorithm + b'foobar') # certOrCRL + + + @classmethod + def OBJECT(cls): + """ + @return: A L{dns.Record_CERT} instance with attributes that + match the encoded record returned by L{BYTES}. + """ + return dns.Record_CERT( + certType=2, + keyTag=1, + algorithm=4, + certOrCRL=b'foobar') + + + +class CERTRecordTests(unittest.TestCase): + """ + Tests for L{dns.Record_CERT}. + """ + + def test_certTypeDefaultAttribute(self): + """ + L{dns.Record_CERT.certType} is a public L{int} attribute + encoding the type of key the record holds. The + default value is C{1}. + + http://tools.ietf.org/html/rfc4398#section-2.1 + https://www.iana.org/assignments/cert-rr-types/cert-rr-types.xhtml + """ + record = dns.Record_CERT() + self.assertEqual(record.certType, 1) + + + def test_certTypeOverride(self): + """ + L{dns.Record_CERT.__init__} accepts a C{certType} parameter + which overrides the L{dns.Record_CERT.certType} attribute. + """ + record = dns.Record_CERT(certType=2) + self.assertEqual(record.certType, 2) + + + def test_keyTagDefaultAttribute(self): + """ + L{dns.Record_CERT.keyTag} is a public L{int} + attribute holding an identifier generated from the certificate. + The default is C{0}. + """ + record = dns.Record_CERT() + self.assertEqual(record.keyTag, 0) + + + def test_keyTagOverride(self): + """ + L{dns.Record_CERT.__init__} accepts a C{keyTag} + parameter which overrides the + L{dns.Record_CERT.keyTag} attribute. + """ + record = dns.Record_CERT(keyTag=1) + self.assertEqual(record.keyTag, 1) + + + def test_algorithmDefaultAttribute(self): + """ + L{dns.Record_CERT.algorithm} is a public L{int} attribute + whose default value is 5 (RSA/SHA-1). + + Values are defined in DNSSEC + https://tools.ietf.org/html/rfc4034#section-2.1.3 + """ + record = dns.Record_CERT() + self.assertEqual(record.algorithm, 5) + + + def test_algorithmOverride(self): + """ + L{dns.Record_CERT.__init__} accepts a C{algorithm} + parameter which overrides the + L{dns.Record_CERT.algorithm} attribute. + """ + record = dns.Record_CERT(algorithm=255) + self.assertEqual(record.algorithm, 255) + + + def test_certOrCRLDefaultAttribute(self): + """ + L{dns.Record_CERT.certOrCRL} is a public L{bytes} attribute + whose default value is C{b''}. + """ + record = dns.Record_CERT() + self.assertEqual(record.certOrCRL, b'') + + + def test_certOrCRLOverride(self): + """ + L{dns.Record_CERT.__init__} accepts a C{certOrCRL} + parameter which overrides the + L{dns.Record_CERT.certOrCRL} attribute. + """ + record = dns.Record_CERT(certOrCRL=b'foobar') + self.assertEqual(record.certOrCRL, b'foobar') + + + def test_encode(self): + """ + L{dns.Record_CERT.encode} packs the header fields and the + key and writes them to a file like object passed in as an + argument. + """ + record = CERT_TEST_DATA.OBJECT() + actualBytes = BytesIO() + record.encode(actualBytes) + + self.assertEqual(actualBytes.getvalue(), CERT_TEST_DATA.BYTES()) + + + def test_decode(self): + """ + L{dns.Record_CERT.decode} unpacks the header fields from a file + like object and populates the attributes of an existing + L{dns.Record_CERT} instance. + """ + expectedBytes = CERT_TEST_DATA.BYTES() + record = dns.Record_CERT() + record.decode(BytesIO(expectedBytes), length=len(expectedBytes)) + + self.assertEqual(record, CERT_TEST_DATA.OBJECT()) + + + def test_decodeShorterThanHeader(self): + """ + L{dns.Record_CERT.decode} raises L{EOFError} if the provided + file object is shorter than the fixed length header parts. ie + everything except key. + """ + record = dns.Record_CERT() + + self.assertRaises(EOFError, record.decode, BytesIO(b'x'), length=1) + + + def test_decodeShorterThanKey(self): + """ + L{dns.Record_CERT.decode} raises L{EOFError} if the provided + file object is shorter than length provided in the length + argument. + """ + expectedBytes = CERT_TEST_DATA.BYTES() + record = dns.Record_CERT() + + self.assertRaises( + EOFError, + record.decode, + BytesIO(expectedBytes[:-1]), length=len(expectedBytes)) Index: twisted/names/test/test_names.py =================================================================== --- twisted/names/test/test_names.py (revision 40421) +++ twisted/names/test/test_names.py (working copy) @@ -74,6 +74,7 @@ dns.Record_NS('39.28.189.39'), dns.Record_SPF('v=spf1 mx/30 mx:example.org/30 -all'), dns.Record_SPF('v=spf1 +mx a:\0colo', '.example.com/28 -all not valid'), + dns.Record_CERT(certOrCRL=b'\x01\x02'), dns.Record_MX(10, 'host.test-domain.com'), dns.Record_HINFO(os='Linux', cpu='A Fast One, Dontcha know'), dns.Record_CNAME('canonical.name.com'), @@ -361,6 +362,15 @@ ) + def testCERT(self): + """ + L{DNSServerFactory} can serve I{CERT} resource records. + """ + return self.namesTest( + self.resolver.lookupCertificate('test-domain.com'), + [dns.Record_CERT(certOrCRL=b'\x01\x02', ttl=19283784)] + ) + def testWKS(self): """Test DNS 'WKS' record queries""" return self.namesTest(