lib/dnspython/dns/tsig.py

   1 # Copyright (C) 2001-2007, 2009-2011 Nominum, Inc.
   2 #
   3 # Permission to use, copy, modify, and distribute this software and its
   4 # documentation for any purpose with or without fee is hereby granted,
   5 # provided that the above copyright notice and this permission notice
   6 # appear in all copies.
   7 #
   8 # THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
   9 # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  10 # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
  11 # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  12 # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  13 # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
  14 # OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  15
  16 """DNS TSIG support."""
  17
  18 import hmac
  19 import struct
  20 import sys
  21
  22 import dns.exception
  23 import dns.hash
  24 import dns.rdataclass
  25 import dns.name
  26
  27 class BadTime(dns.exception.DNSException):
  28     """Raised if the current time is not within the TSIG's validity time."""
  29     pass
  30
  31 class BadSignature(dns.exception.DNSException):
  32     """Raised if the TSIG signature fails to verify."""
  33     pass
  34
  35 class PeerError(dns.exception.DNSException):
  36     """Base class for all TSIG errors generated by the remote peer"""
  37     pass
  38
  39 class PeerBadKey(PeerError):
  40     """Raised if the peer didn't know the key we used"""
  41     pass
  42
  43 class PeerBadSignature(PeerError):
  44     """Raised if the peer didn't like the signature we sent"""
  45     pass
  46
  47 class PeerBadTime(PeerError):
  48     """Raised if the peer didn't like the time we sent"""
  49     pass
  50
  51 class PeerBadTruncation(PeerError):
  52     """Raised if the peer didn't like amount of truncation in the TSIG we sent"""
  53     pass
  54
  55 # TSIG Algorithms
  56
  57 HMAC_MD5 = dns.name.from_text("HMAC-MD5.SIG-ALG.REG.INT")
  58 HMAC_SHA1 = dns.name.from_text("hmac-sha1")
  59 HMAC_SHA224 = dns.name.from_text("hmac-sha224")
  60 HMAC_SHA256 = dns.name.from_text("hmac-sha256")
  61 HMAC_SHA384 = dns.name.from_text("hmac-sha384")
  62 HMAC_SHA512 = dns.name.from_text("hmac-sha512")
  63
  64 default_algorithm = HMAC_MD5
  65
  66 BADSIG = 16
  67 BADKEY = 17
  68 BADTIME = 18
  69 BADTRUNC = 22
  70
  71 def sign(wire, keyname, secret, time, fudge, original_id, error,
  72          other_data, request_mac, ctx=None, multi=False, first=True,
  73          algorithm=default_algorithm):
  74     """Return a (tsig_rdata, mac, ctx) tuple containing the HMAC TSIG rdata
  75     for the input parameters, the HMAC MAC calculated by applying the
  76     TSIG signature algorithm, and the TSIG digest context.
  77     @rtype: (string, string, hmac.HMAC object)
  78     @raises ValueError: I{other_data} is too long
  79     @raises NotImplementedError: I{algorithm} is not supported
  80     """
  81
  82     (algorithm_name, digestmod) = get_algorithm(algorithm)
  83     if first:
  84         ctx = hmac.new(secret, digestmod=digestmod)
  85         ml = len(request_mac)
  86         if ml > 0:
  87             ctx.update(struct.pack('!H', ml))
  88             ctx.update(request_mac)
  89     id = struct.pack('!H', original_id)
  90     ctx.update(id)
  91     ctx.update(wire[2:])
  92     if first:
  93         ctx.update(keyname.to_digestable())
  94         ctx.update(struct.pack('!H', dns.rdataclass.ANY))
  95         ctx.update(struct.pack('!I', 0))
  96     long_time = time + 0L
  97     upper_time = (long_time >> 32) & 0xffffL
  98     lower_time = long_time & 0xffffffffL
  99     time_mac = struct.pack('!HIH', upper_time, lower_time, fudge)
 100     pre_mac = algorithm_name + time_mac
 101     ol = len(other_data)
 102     if ol > 65535:
 103         raise ValueError('TSIG Other Data is > 65535 bytes')
 104     post_mac = struct.pack('!HH', error, ol) + other_data
 105     if first:
 106         ctx.update(pre_mac)
 107         ctx.update(post_mac)
 108     else:
 109         ctx.update(time_mac)
 110     mac = ctx.digest()
 111     mpack = struct.pack('!H', len(mac))
 112     tsig_rdata = pre_mac + mpack + mac + id + post_mac
 113     if multi:
 114         ctx = hmac.new(secret)
 115         ml = len(mac)
 116         ctx.update(struct.pack('!H', ml))
 117         ctx.update(mac)
 118     else:
 119         ctx = None
 120     return (tsig_rdata, mac, ctx)
 121
 122 def hmac_md5(wire, keyname, secret, time, fudge, original_id, error,
 123              other_data, request_mac, ctx=None, multi=False, first=True,
 124              algorithm=default_algorithm):
 125     return sign(wire, keyname, secret, time, fudge, original_id, error,
 126                 other_data, request_mac, ctx, multi, first, algorithm)
 127
 128 def validate(wire, keyname, secret, now, request_mac, tsig_start, tsig_rdata,
 129              tsig_rdlen, ctx=None, multi=False, first=True):
 130     """Validate the specified TSIG rdata against the other input parameters.
 131
 132     @raises FormError: The TSIG is badly formed.
 133     @raises BadTime: There is too much time skew between the client and the
 134     server.
 135     @raises BadSignature: The TSIG signature did not validate
 136     @rtype: hmac.HMAC object"""
 137
 138     (adcount,) = struct.unpack("!H", wire[10:12])
 139     if adcount == 0:
 140         raise dns.exception.FormError
 141     adcount -= 1
 142     new_wire = wire[0:10] + struct.pack("!H", adcount) + wire[12:tsig_start]
 143     current = tsig_rdata
 144     (aname, used) = dns.name.from_wire(wire, current)
 145     current = current + used
 146     (upper_time, lower_time, fudge, mac_size) = \
 147                  struct.unpack("!HIHH", wire[current:current + 10])
 148     time = ((upper_time + 0L) << 32) + (lower_time + 0L)
 149     current += 10
 150     mac = wire[current:current + mac_size]
 151     current += mac_size
 152     (original_id, error, other_size) = \
 153                   struct.unpack("!HHH", wire[current:current + 6])
 154     current += 6
 155     other_data = wire[current:current + other_size]
 156     current += other_size
 157     if current != tsig_rdata + tsig_rdlen:
 158         raise dns.exception.FormError
 159     if error != 0:
 160         if error == BADSIG:
 161             raise PeerBadSignature
 162         elif error == BADKEY:
 163             raise PeerBadKey
 164         elif error == BADTIME:
 165             raise PeerBadTime
 166         elif error == BADTRUNC:
 167             raise PeerBadTruncation
 168         else:
 169             raise PeerError('unknown TSIG error code %d' % error)
 170     time_low = time - fudge
 171     time_high = time + fudge
 172     if now < time_low or now > time_high:
 173         raise BadTime
 174     (junk, our_mac, ctx) = sign(new_wire, keyname, secret, time, fudge,
 175                                 original_id, error, other_data,
 176                                 request_mac, ctx, multi, first, aname)
 177     if (our_mac != mac):
 178         raise BadSignature
 179     return ctx
 180
 181 _hashes = None
 182
 183 def _maybe_add_hash(tsig_alg, hash_alg):
 184     try:
 185         _hashes[tsig_alg] = dns.hash.get(hash_alg)
 186     except KeyError:
 187         pass
 188
 189 def _setup_hashes():
 190     global _hashes
 191     _hashes = {}
 192     _maybe_add_hash(HMAC_SHA224, 'SHA224')
 193     _maybe_add_hash(HMAC_SHA256, 'SHA256')
 194     _maybe_add_hash(HMAC_SHA384, 'SHA384')
 195     _maybe_add_hash(HMAC_SHA512, 'SHA512')
 196     _maybe_add_hash(HMAC_SHA1, 'SHA1')
 197     _maybe_add_hash(HMAC_MD5, 'MD5')
 198
 199 def get_algorithm(algorithm):
 200     """Returns the wire format string and the hash module to use for the
 201     specified TSIG algorithm
 202
 203     @rtype: (string, hash constructor)
 204     @raises NotImplementedError: I{algorithm} is not supported
 205     """
 206
 207     global _hashes
 208     if _hashes is None:
 209         _setup_hashes()
 210
 211     if isinstance(algorithm, (str, unicode)):
 212         algorithm = dns.name.from_text(algorithm)
 213
 214     if sys.hexversion < 0x02050200 and \
 215        (algorithm == HMAC_SHA384 or algorithm == HMAC_SHA512):
 216         raise NotImplementedError("TSIG algorithm " + str(algorithm) +
 217                                   " requires Python 2.5.2 or later")
 218
 219     try:
 220         return (algorithm.to_digestable(), _hashes[algorithm])
 221     except KeyError:
 222         raise NotImplementedError("TSIG algorithm " + str(algorithm) +
 223                                   " is not supported")