diff options
Diffstat (limited to 'lib/dnspython/dns/tsig.py')
-rw-r--r-- | lib/dnspython/dns/tsig.py | 216 |
1 files changed, 216 insertions, 0 deletions
diff --git a/lib/dnspython/dns/tsig.py b/lib/dnspython/dns/tsig.py new file mode 100644 index 0000000000..b4deeca859 --- /dev/null +++ b/lib/dnspython/dns/tsig.py @@ -0,0 +1,216 @@ +# Copyright (C) 2001-2007, 2009, 2010 Nominum, Inc. +# +# Permission to use, copy, modify, and distribute this software and its +# documentation for any purpose with or without fee is hereby granted, +# provided that the above copyright notice and this permission notice +# appear in all copies. +# +# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES +# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF +# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR +# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES +# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN +# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT +# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + +"""DNS TSIG support.""" + +import hmac +import struct + +import dns.exception +import dns.rdataclass +import dns.name + +class BadTime(dns.exception.DNSException): + """Raised if the current time is not within the TSIG's validity time.""" + pass + +class BadSignature(dns.exception.DNSException): + """Raised if the TSIG signature fails to verify.""" + pass + +class PeerError(dns.exception.DNSException): + """Base class for all TSIG errors generated by the remote peer""" + pass + +class PeerBadKey(PeerError): + """Raised if the peer didn't know the key we used""" + pass + +class PeerBadSignature(PeerError): + """Raised if the peer didn't like the signature we sent""" + pass + +class PeerBadTime(PeerError): + """Raised if the peer didn't like the time we sent""" + pass + +class PeerBadTruncation(PeerError): + """Raised if the peer didn't like amount of truncation in the TSIG we sent""" + pass + +default_algorithm = "HMAC-MD5.SIG-ALG.REG.INT" + +BADSIG = 16 +BADKEY = 17 +BADTIME = 18 +BADTRUNC = 22 + +def sign(wire, keyname, secret, time, fudge, original_id, error, + other_data, request_mac, ctx=None, multi=False, first=True, + algorithm=default_algorithm): + """Return a (tsig_rdata, mac, ctx) tuple containing the HMAC TSIG rdata + for the input parameters, the HMAC MAC calculated by applying the + TSIG signature algorithm, and the TSIG digest context. + @rtype: (string, string, hmac.HMAC object) + @raises ValueError: I{other_data} is too long + @raises NotImplementedError: I{algorithm} is not supported + """ + + (algorithm_name, digestmod) = get_algorithm(algorithm) + if first: + ctx = hmac.new(secret, digestmod=digestmod) + ml = len(request_mac) + if ml > 0: + ctx.update(struct.pack('!H', ml)) + ctx.update(request_mac) + id = struct.pack('!H', original_id) + ctx.update(id) + ctx.update(wire[2:]) + if first: + ctx.update(keyname.to_digestable()) + ctx.update(struct.pack('!H', dns.rdataclass.ANY)) + ctx.update(struct.pack('!I', 0)) + long_time = time + 0L + upper_time = (long_time >> 32) & 0xffffL + lower_time = long_time & 0xffffffffL + time_mac = struct.pack('!HIH', upper_time, lower_time, fudge) + pre_mac = algorithm_name + time_mac + ol = len(other_data) + if ol > 65535: + raise ValueError('TSIG Other Data is > 65535 bytes') + post_mac = struct.pack('!HH', error, ol) + other_data + if first: + ctx.update(pre_mac) + ctx.update(post_mac) + else: + ctx.update(time_mac) + mac = ctx.digest() + mpack = struct.pack('!H', len(mac)) + tsig_rdata = pre_mac + mpack + mac + id + post_mac + if multi: + ctx = hmac.new(secret) + ml = len(mac) + ctx.update(struct.pack('!H', ml)) + ctx.update(mac) + else: + ctx = None + return (tsig_rdata, mac, ctx) + +def hmac_md5(wire, keyname, secret, time, fudge, original_id, error, + other_data, request_mac, ctx=None, multi=False, first=True, + algorithm=default_algorithm): + return sign(wire, keyname, secret, time, fudge, original_id, error, + other_data, request_mac, ctx, multi, first, algorithm) + +def validate(wire, keyname, secret, now, request_mac, tsig_start, tsig_rdata, + tsig_rdlen, ctx=None, multi=False, first=True): + """Validate the specified TSIG rdata against the other input parameters. + + @raises FormError: The TSIG is badly formed. + @raises BadTime: There is too much time skew between the client and the + server. + @raises BadSignature: The TSIG signature did not validate + @rtype: hmac.HMAC object""" + + (adcount,) = struct.unpack("!H", wire[10:12]) + if adcount == 0: + raise dns.exception.FormError + adcount -= 1 + new_wire = wire[0:10] + struct.pack("!H", adcount) + wire[12:tsig_start] + current = tsig_rdata + (aname, used) = dns.name.from_wire(wire, current) + current = current + used + (upper_time, lower_time, fudge, mac_size) = \ + struct.unpack("!HIHH", wire[current:current + 10]) + time = ((upper_time + 0L) << 32) + (lower_time + 0L) + current += 10 + mac = wire[current:current + mac_size] + current += mac_size + (original_id, error, other_size) = \ + struct.unpack("!HHH", wire[current:current + 6]) + current += 6 + other_data = wire[current:current + other_size] + current += other_size + if current != tsig_rdata + tsig_rdlen: + raise dns.exception.FormError + if error != 0: + if error == BADSIG: + raise PeerBadSignature + elif error == BADKEY: + raise PeerBadKey + elif error == BADTIME: + raise PeerBadTime + elif error == BADTRUNC: + raise PeerBadTruncation + else: + raise PeerError('unknown TSIG error code %d' % error) + time_low = time - fudge + time_high = time + fudge + if now < time_low or now > time_high: + raise BadTime + (junk, our_mac, ctx) = sign(new_wire, keyname, secret, time, fudge, + original_id, error, other_data, + request_mac, ctx, multi, first, aname) + if (our_mac != mac): + raise BadSignature + return ctx + +def get_algorithm(algorithm): + """Returns the wire format string and the hash module to use for the + specified TSIG algorithm + + @rtype: (string, hash constructor) + @raises NotImplementedError: I{algorithm} is not supported + """ + + hashes = {} + try: + import hashlib + hashes[dns.name.from_text('hmac-sha224')] = hashlib.sha224 + hashes[dns.name.from_text('hmac-sha256')] = hashlib.sha256 + hashes[dns.name.from_text('hmac-sha384')] = hashlib.sha384 + hashes[dns.name.from_text('hmac-sha512')] = hashlib.sha512 + hashes[dns.name.from_text('hmac-sha1')] = hashlib.sha1 + hashes[dns.name.from_text('HMAC-MD5.SIG-ALG.REG.INT')] = hashlib.md5 + + import sys + if sys.hexversion < 0x02050000: + # hashlib doesn't conform to PEP 247: API for + # Cryptographic Hash Functions, which hmac before python + # 2.5 requires, so add the necessary items. + class HashlibWrapper: + def __init__(self, basehash): + self.basehash = basehash + self.digest_size = self.basehash().digest_size + + def new(self, *args, **kwargs): + return self.basehash(*args, **kwargs) + + for name in hashes: + hashes[name] = HashlibWrapper(hashes[name]) + + except ImportError: + import md5, sha + hashes[dns.name.from_text('HMAC-MD5.SIG-ALG.REG.INT')] = md5.md5 + hashes[dns.name.from_text('hmac-sha1')] = sha.sha + + if isinstance(algorithm, (str, unicode)): + algorithm = dns.name.from_text(algorithm) + + if algorithm in hashes: + return (algorithm.to_digestable(), hashes[algorithm]) + + raise NotImplementedError("TSIG algorithm " + str(algorithm) + + " is not supported") |