tsig.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352
  1. # Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license
  2. # Copyright (C) 2001-2007, 2009-2011 Nominum, Inc.
  3. #
  4. # Permission to use, copy, modify, and distribute this software and its
  5. # documentation for any purpose with or without fee is hereby granted,
  6. # provided that the above copyright notice and this permission notice
  7. # appear in all copies.
  8. #
  9. # THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES
  10. # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  11. # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR
  12. # ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  13. # WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  14. # ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT
  15. # OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  16. """DNS TSIG support."""
  17. import base64
  18. import hashlib
  19. import hmac
  20. import struct
  21. import dns.exception
  22. import dns.name
  23. import dns.rcode
  24. import dns.rdataclass
  25. class BadTime(dns.exception.DNSException):
  26. """The current time is not within the TSIG's validity time."""
  27. class BadSignature(dns.exception.DNSException):
  28. """The TSIG signature fails to verify."""
  29. class BadKey(dns.exception.DNSException):
  30. """The TSIG record owner name does not match the key."""
  31. class BadAlgorithm(dns.exception.DNSException):
  32. """The TSIG algorithm does not match the key."""
  33. class PeerError(dns.exception.DNSException):
  34. """Base class for all TSIG errors generated by the remote peer"""
  35. class PeerBadKey(PeerError):
  36. """The peer didn't know the key we used"""
  37. class PeerBadSignature(PeerError):
  38. """The peer didn't like the signature we sent"""
  39. class PeerBadTime(PeerError):
  40. """The peer didn't like the time we sent"""
  41. class PeerBadTruncation(PeerError):
  42. """The peer didn't like amount of truncation in the TSIG we sent"""
  43. # TSIG Algorithms
  44. HMAC_MD5 = dns.name.from_text("HMAC-MD5.SIG-ALG.REG.INT")
  45. HMAC_SHA1 = dns.name.from_text("hmac-sha1")
  46. HMAC_SHA224 = dns.name.from_text("hmac-sha224")
  47. HMAC_SHA256 = dns.name.from_text("hmac-sha256")
  48. HMAC_SHA256_128 = dns.name.from_text("hmac-sha256-128")
  49. HMAC_SHA384 = dns.name.from_text("hmac-sha384")
  50. HMAC_SHA384_192 = dns.name.from_text("hmac-sha384-192")
  51. HMAC_SHA512 = dns.name.from_text("hmac-sha512")
  52. HMAC_SHA512_256 = dns.name.from_text("hmac-sha512-256")
  53. GSS_TSIG = dns.name.from_text("gss-tsig")
  54. default_algorithm = HMAC_SHA256
  55. mac_sizes = {
  56. HMAC_SHA1: 20,
  57. HMAC_SHA224: 28,
  58. HMAC_SHA256: 32,
  59. HMAC_SHA256_128: 16,
  60. HMAC_SHA384: 48,
  61. HMAC_SHA384_192: 24,
  62. HMAC_SHA512: 64,
  63. HMAC_SHA512_256: 32,
  64. HMAC_MD5: 16,
  65. GSS_TSIG: 128, # This is what we assume to be the worst case!
  66. }
  67. class GSSTSig:
  68. """
  69. GSS-TSIG TSIG implementation. This uses the GSS-API context established
  70. in the TKEY message handshake to sign messages using GSS-API message
  71. integrity codes, per the RFC.
  72. In order to avoid a direct GSSAPI dependency, the keyring holds a ref
  73. to the GSSAPI object required, rather than the key itself.
  74. """
  75. def __init__(self, gssapi_context):
  76. self.gssapi_context = gssapi_context
  77. self.data = b""
  78. self.name = "gss-tsig"
  79. def update(self, data):
  80. self.data += data
  81. def sign(self):
  82. # defer to the GSSAPI function to sign
  83. return self.gssapi_context.get_signature(self.data)
  84. def verify(self, expected):
  85. try:
  86. # defer to the GSSAPI function to verify
  87. return self.gssapi_context.verify_signature(self.data, expected)
  88. except Exception:
  89. # note the usage of a bare exception
  90. raise BadSignature
  91. class GSSTSigAdapter:
  92. def __init__(self, keyring):
  93. self.keyring = keyring
  94. def __call__(self, message, keyname):
  95. if keyname in self.keyring:
  96. key = self.keyring[keyname]
  97. if isinstance(key, Key) and key.algorithm == GSS_TSIG:
  98. if message:
  99. GSSTSigAdapter.parse_tkey_and_step(key, message, keyname)
  100. return key
  101. else:
  102. return None
  103. @classmethod
  104. def parse_tkey_and_step(cls, key, message, keyname):
  105. # if the message is a TKEY type, absorb the key material
  106. # into the context using step(); this is used to allow the
  107. # client to complete the GSSAPI negotiation before attempting
  108. # to verify the signed response to a TKEY message exchange
  109. try:
  110. rrset = message.find_rrset(
  111. message.answer, keyname, dns.rdataclass.ANY, dns.rdatatype.TKEY
  112. )
  113. if rrset:
  114. token = rrset[0].key
  115. gssapi_context = key.secret
  116. return gssapi_context.step(token)
  117. except KeyError:
  118. pass
  119. class HMACTSig:
  120. """
  121. HMAC TSIG implementation. This uses the HMAC python module to handle the
  122. sign/verify operations.
  123. """
  124. _hashes = {
  125. HMAC_SHA1: hashlib.sha1,
  126. HMAC_SHA224: hashlib.sha224,
  127. HMAC_SHA256: hashlib.sha256,
  128. HMAC_SHA256_128: (hashlib.sha256, 128),
  129. HMAC_SHA384: hashlib.sha384,
  130. HMAC_SHA384_192: (hashlib.sha384, 192),
  131. HMAC_SHA512: hashlib.sha512,
  132. HMAC_SHA512_256: (hashlib.sha512, 256),
  133. HMAC_MD5: hashlib.md5,
  134. }
  135. def __init__(self, key, algorithm):
  136. try:
  137. hashinfo = self._hashes[algorithm]
  138. except KeyError:
  139. raise NotImplementedError(f"TSIG algorithm {algorithm} is not supported")
  140. # create the HMAC context
  141. if isinstance(hashinfo, tuple):
  142. self.hmac_context = hmac.new(key, digestmod=hashinfo[0])
  143. self.size = hashinfo[1]
  144. else:
  145. self.hmac_context = hmac.new(key, digestmod=hashinfo)
  146. self.size = None
  147. self.name = self.hmac_context.name
  148. if self.size:
  149. self.name += f"-{self.size}"
  150. def update(self, data):
  151. return self.hmac_context.update(data)
  152. def sign(self):
  153. # defer to the HMAC digest() function for that digestmod
  154. digest = self.hmac_context.digest()
  155. if self.size:
  156. digest = digest[: (self.size // 8)]
  157. return digest
  158. def verify(self, expected):
  159. # re-digest and compare the results
  160. mac = self.sign()
  161. if not hmac.compare_digest(mac, expected):
  162. raise BadSignature
  163. def _digest(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=None):
  164. """Return a context containing the TSIG rdata for the input parameters
  165. @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object
  166. @raises ValueError: I{other_data} is too long
  167. @raises NotImplementedError: I{algorithm} is not supported
  168. """
  169. first = not (ctx and multi)
  170. if first:
  171. ctx = get_context(key)
  172. if request_mac:
  173. ctx.update(struct.pack("!H", len(request_mac)))
  174. ctx.update(request_mac)
  175. ctx.update(struct.pack("!H", rdata.original_id))
  176. ctx.update(wire[2:])
  177. if first:
  178. ctx.update(key.name.to_digestable())
  179. ctx.update(struct.pack("!H", dns.rdataclass.ANY))
  180. ctx.update(struct.pack("!I", 0))
  181. if time is None:
  182. time = rdata.time_signed
  183. upper_time = (time >> 32) & 0xFFFF
  184. lower_time = time & 0xFFFFFFFF
  185. time_encoded = struct.pack("!HIH", upper_time, lower_time, rdata.fudge)
  186. other_len = len(rdata.other)
  187. if other_len > 65535:
  188. raise ValueError("TSIG Other Data is > 65535 bytes")
  189. if first:
  190. ctx.update(key.algorithm.to_digestable() + time_encoded)
  191. ctx.update(struct.pack("!HH", rdata.error, other_len) + rdata.other)
  192. else:
  193. ctx.update(time_encoded)
  194. return ctx
  195. def _maybe_start_digest(key, mac, multi):
  196. """If this is the first message in a multi-message sequence,
  197. start a new context.
  198. @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object
  199. """
  200. if multi:
  201. ctx = get_context(key)
  202. ctx.update(struct.pack("!H", len(mac)))
  203. ctx.update(mac)
  204. return ctx
  205. else:
  206. return None
  207. def sign(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=False):
  208. """Return a (tsig_rdata, mac, ctx) tuple containing the HMAC TSIG rdata
  209. for the input parameters, the HMAC MAC calculated by applying the
  210. TSIG signature algorithm, and the TSIG digest context.
  211. @rtype: (string, dns.tsig.HMACTSig or dns.tsig.GSSTSig object)
  212. @raises ValueError: I{other_data} is too long
  213. @raises NotImplementedError: I{algorithm} is not supported
  214. """
  215. ctx = _digest(wire, key, rdata, time, request_mac, ctx, multi)
  216. mac = ctx.sign()
  217. tsig = rdata.replace(time_signed=time, mac=mac)
  218. return (tsig, _maybe_start_digest(key, mac, multi))
  219. def validate(
  220. wire, key, owner, rdata, now, request_mac, tsig_start, ctx=None, multi=False
  221. ):
  222. """Validate the specified TSIG rdata against the other input parameters.
  223. @raises FormError: The TSIG is badly formed.
  224. @raises BadTime: There is too much time skew between the client and the
  225. server.
  226. @raises BadSignature: The TSIG signature did not validate
  227. @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object"""
  228. (adcount,) = struct.unpack("!H", wire[10:12])
  229. if adcount == 0:
  230. raise dns.exception.FormError
  231. adcount -= 1
  232. new_wire = wire[0:10] + struct.pack("!H", adcount) + wire[12:tsig_start]
  233. if rdata.error != 0:
  234. if rdata.error == dns.rcode.BADSIG:
  235. raise PeerBadSignature
  236. elif rdata.error == dns.rcode.BADKEY:
  237. raise PeerBadKey
  238. elif rdata.error == dns.rcode.BADTIME:
  239. raise PeerBadTime
  240. elif rdata.error == dns.rcode.BADTRUNC:
  241. raise PeerBadTruncation
  242. else:
  243. raise PeerError("unknown TSIG error code %d" % rdata.error)
  244. if abs(rdata.time_signed - now) > rdata.fudge:
  245. raise BadTime
  246. if key.name != owner:
  247. raise BadKey
  248. if key.algorithm != rdata.algorithm:
  249. raise BadAlgorithm
  250. ctx = _digest(new_wire, key, rdata, None, request_mac, ctx, multi)
  251. ctx.verify(rdata.mac)
  252. return _maybe_start_digest(key, rdata.mac, multi)
  253. def get_context(key):
  254. """Returns an HMAC context for the specified key.
  255. @rtype: HMAC context
  256. @raises NotImplementedError: I{algorithm} is not supported
  257. """
  258. if key.algorithm == GSS_TSIG:
  259. return GSSTSig(key.secret)
  260. else:
  261. return HMACTSig(key.secret, key.algorithm)
  262. class Key:
  263. def __init__(self, name, secret, algorithm=default_algorithm):
  264. if isinstance(name, str):
  265. name = dns.name.from_text(name)
  266. self.name = name
  267. if isinstance(secret, str):
  268. secret = base64.decodebytes(secret.encode())
  269. self.secret = secret
  270. if isinstance(algorithm, str):
  271. algorithm = dns.name.from_text(algorithm)
  272. self.algorithm = algorithm
  273. def __eq__(self, other):
  274. return (
  275. isinstance(other, Key)
  276. and self.name == other.name
  277. and self.secret == other.secret
  278. and self.algorithm == other.algorithm
  279. )
  280. def __repr__(self):
  281. r = f"<DNS key name='{self.name}', " + f"algorithm='{self.algorithm}'"
  282. if self.algorithm != GSS_TSIG:
  283. r += f", secret='{base64.b64encode(self.secret).decode()}'"
  284. r += ">"
  285. return r