Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dns/tsig.py: 31%

1# Copyright (C) Dnspython Contributors, see LICENSE for text of ISC license 

2 

3# Copyright (C) 2001-2007, 2009-2011 Nominum, Inc. 

4# 

5# Permission to use, copy, modify, and distribute this software and its 

6# documentation for any purpose with or without fee is hereby granted, 

7# provided that the above copyright notice and this permission notice 

8# appear in all copies. 

9# 

10# THE SOFTWARE IS PROVIDED "AS IS" AND NOMINUM DISCLAIMS ALL WARRANTIES 

11# WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF 

12# MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL NOMINUM BE LIABLE FOR 

13# ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES 

14# WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN 

15# ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT 

16# OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. 

17 

18"""DNS TSIG support.""" 

19 

20import base64 

21import hashlib 

22import hmac 

23import struct 

24 

25import dns.exception 

26import dns.name 

27import dns.rcode 

28import dns.rdataclass 

29import dns.rdatatype 

30 

31 

32class BadTime(dns.exception.DNSException): 

33 """The current time is not within the TSIG's validity time.""" 

34 

35 

36class BadSignature(dns.exception.DNSException): 

37 """The TSIG signature fails to verify.""" 

38 

39 

40class BadKey(dns.exception.DNSException): 

41 """The TSIG record owner name does not match the key.""" 

42 

43 

44class BadAlgorithm(dns.exception.DNSException): 

45 """The TSIG algorithm does not match the key.""" 

46 

47 

48class PeerError(dns.exception.DNSException): 

49 """Base class for all TSIG errors generated by the remote peer""" 

50 

51 

52class PeerBadKey(PeerError): 

53 """The peer didn't know the key we used""" 

54 

55 

56class PeerBadSignature(PeerError): 

57 """The peer didn't like the signature we sent""" 

58 

59 

60class PeerBadTime(PeerError): 

61 """The peer didn't like the time we sent""" 

62 

63 

64class PeerBadTruncation(PeerError): 

65 """The peer didn't like amount of truncation in the TSIG we sent""" 

66 

67 

68# TSIG Algorithms 

69 

70HMAC_MD5 = dns.name.from_text("HMAC-MD5.SIG-ALG.REG.INT") 

71HMAC_SHA1 = dns.name.from_text("hmac-sha1") 

72HMAC_SHA224 = dns.name.from_text("hmac-sha224") 

73HMAC_SHA256 = dns.name.from_text("hmac-sha256") 

74HMAC_SHA256_128 = dns.name.from_text("hmac-sha256-128") 

75HMAC_SHA384 = dns.name.from_text("hmac-sha384") 

76HMAC_SHA384_192 = dns.name.from_text("hmac-sha384-192") 

77HMAC_SHA512 = dns.name.from_text("hmac-sha512") 

78HMAC_SHA512_256 = dns.name.from_text("hmac-sha512-256") 

79GSS_TSIG = dns.name.from_text("gss-tsig") 

80 

81default_algorithm = HMAC_SHA256 

82 

83mac_sizes = { 

84 HMAC_SHA1: 20, 

85 HMAC_SHA224: 28, 

86 HMAC_SHA256: 32, 

87 HMAC_SHA256_128: 16, 

88 HMAC_SHA384: 48, 

89 HMAC_SHA384_192: 24, 

90 HMAC_SHA512: 64, 

91 HMAC_SHA512_256: 32, 

92 HMAC_MD5: 16, 

93 GSS_TSIG: 128, # This is what we assume to be the worst case! 

94} 

95 

96 

97class GSSTSig: 

98 """ 

99 GSS-TSIG TSIG implementation. This uses the GSS-API context established 

100 in the TKEY message handshake to sign messages using GSS-API message 

101 integrity codes, per the RFC. 

102 

103 In order to avoid a direct GSSAPI dependency, the keyring holds a ref 

104 to the GSSAPI object required, rather than the key itself. 

105 """ 

106 

107 def __init__(self, gssapi_context): 

108 self.gssapi_context = gssapi_context 

109 self.data = b"" 

110 self.name = "gss-tsig" 

111 

112 def update(self, data): 

113 self.data += data 

114 

115 def sign(self): 

116 # defer to the GSSAPI function to sign 

117 return self.gssapi_context.get_signature(self.data) 

118 

119 def verify(self, expected): 

120 try: 

121 # defer to the GSSAPI function to verify 

122 return self.gssapi_context.verify_signature(self.data, expected) 

123 except Exception: 

124 # note the usage of a bare exception 

125 raise BadSignature 

126 

127 

128class GSSTSigAdapter: 

129 def __init__(self, keyring): 

130 self.keyring = keyring 

131 

132 def __call__(self, message, keyname): 

133 if keyname in self.keyring: 

134 key = self.keyring[keyname] 

135 if isinstance(key, Key) and key.algorithm == GSS_TSIG: 

136 if message: 

137 GSSTSigAdapter.parse_tkey_and_step(key, message, keyname) 

138 return key 

139 else: 

140 return None 

141 

142 @classmethod 

143 def parse_tkey_and_step(cls, key, message, keyname): 

144 # if the message is a TKEY type, absorb the key material 

145 # into the context using step(); this is used to allow the 

146 # client to complete the GSSAPI negotiation before attempting 

147 # to verify the signed response to a TKEY message exchange 

148 try: 

149 rrset = message.find_rrset( 

150 message.answer, keyname, dns.rdataclass.ANY, dns.rdatatype.TKEY 

151 ) 

152 if rrset: 

153 token = rrset[0].key 

154 gssapi_context = key.secret 

155 return gssapi_context.step(token) 

156 except KeyError: 

157 pass 

158 

159 

160class HMACTSig: 

161 """ 

162 HMAC TSIG implementation. This uses the HMAC python module to handle the 

163 sign/verify operations. 

164 """ 

165 

166 _hashes = { 

167 HMAC_SHA1: hashlib.sha1, 

168 HMAC_SHA224: hashlib.sha224, 

169 HMAC_SHA256: hashlib.sha256, 

170 HMAC_SHA256_128: (hashlib.sha256, 128), 

171 HMAC_SHA384: hashlib.sha384, 

172 HMAC_SHA384_192: (hashlib.sha384, 192), 

173 HMAC_SHA512: hashlib.sha512, 

174 HMAC_SHA512_256: (hashlib.sha512, 256), 

175 HMAC_MD5: hashlib.md5, 

176 } 

177 

178 def __init__(self, key, algorithm): 

179 try: 

180 hashinfo = self._hashes[algorithm] 

181 except KeyError: 

182 raise NotImplementedError(f"TSIG algorithm {algorithm} is not supported") 

183 

184 # create the HMAC context 

185 if isinstance(hashinfo, tuple): 

186 self.hmac_context = hmac.new(key, digestmod=hashinfo[0]) 

187 self.size = hashinfo[1] 

188 else: 

189 self.hmac_context = hmac.new(key, digestmod=hashinfo) 

190 self.size = None 

191 self.name = self.hmac_context.name 

192 if self.size: 

193 self.name += f"-{self.size}" 

194 

195 def update(self, data): 

196 return self.hmac_context.update(data) 

197 

198 def sign(self): 

199 # defer to the HMAC digest() function for that digestmod 

200 digest = self.hmac_context.digest() 

201 if self.size: 

202 assert isinstance(self.size, int) 

203 digest = digest[: (self.size // 8)] 

204 return digest 

205 

206 def verify(self, expected): 

207 # re-digest and compare the results 

208 mac = self.sign() 

209 if not hmac.compare_digest(mac, expected): 

210 raise BadSignature 

211 

212 

213def _digest(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=None): 

214 """Return a context containing the TSIG rdata for the input parameters 

215 @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object 

216 @raises ValueError: I{other_data} is too long 

217 @raises NotImplementedError: I{algorithm} is not supported 

218 """ 

219 

220 first = not (ctx and multi) 

221 if first: 

222 ctx = get_context(key) 

223 if request_mac: 

224 ctx.update(struct.pack("!H", len(request_mac))) 

225 ctx.update(request_mac) 

226 assert ctx is not None # for type checkers 

227 ctx.update(struct.pack("!H", rdata.original_id)) 

228 ctx.update(wire[2:]) 

229 if first: 

230 ctx.update(key.name.to_digestable()) 

231 ctx.update(struct.pack("!H", dns.rdataclass.ANY)) 

232 ctx.update(struct.pack("!I", 0)) 

233 if time is None: 

234 time = rdata.time_signed 

235 upper_time = (time >> 32) & 0xFFFF 

236 lower_time = time & 0xFFFFFFFF 

237 time_encoded = struct.pack("!HIH", upper_time, lower_time, rdata.fudge) 

238 other_len = len(rdata.other) 

239 if other_len > 65535: 

240 raise ValueError("TSIG Other Data is > 65535 bytes") 

241 if first: 

242 ctx.update(key.algorithm.to_digestable() + time_encoded) 

243 ctx.update(struct.pack("!HH", rdata.error, other_len) + rdata.other) 

244 else: 

245 ctx.update(time_encoded) 

246 return ctx 

247 

248 

249def _maybe_start_digest(key, mac, multi): 

250 """If this is the first message in a multi-message sequence, 

251 start a new context. 

252 @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object 

253 """ 

254 if multi: 

255 ctx = get_context(key) 

256 ctx.update(struct.pack("!H", len(mac))) 

257 ctx.update(mac) 

258 return ctx 

259 else: 

260 return None 

261 

262 

263def sign(wire, key, rdata, time=None, request_mac=None, ctx=None, multi=False): 

264 """Return a (tsig_rdata, mac, ctx) tuple containing the HMAC TSIG rdata 

265 for the input parameters, the HMAC MAC calculated by applying the 

266 TSIG signature algorithm, and the TSIG digest context. 

267 @rtype: (string, dns.tsig.HMACTSig or dns.tsig.GSSTSig object) 

268 @raises ValueError: I{other_data} is too long 

269 @raises NotImplementedError: I{algorithm} is not supported 

270 """ 

271 

272 ctx = _digest(wire, key, rdata, time, request_mac, ctx, multi) 

273 mac = ctx.sign() 

274 tsig = rdata.replace(time_signed=time, mac=mac) 

275 

276 return (tsig, _maybe_start_digest(key, mac, multi)) 

277 

278 

279def validate( 

280 wire, key, owner, rdata, now, request_mac, tsig_start, ctx=None, multi=False 

281): 

282 """Validate the specified TSIG rdata against the other input parameters. 

283 

284 @raises FormError: The TSIG is badly formed. 

285 @raises BadTime: There is too much time skew between the client and the 

286 server. 

287 @raises BadSignature: The TSIG signature did not validate 

288 @rtype: dns.tsig.HMACTSig or dns.tsig.GSSTSig object""" 

289 

290 (adcount,) = struct.unpack("!H", wire[10:12]) 

291 if adcount == 0: 

292 raise dns.exception.FormError 

293 adcount -= 1 

294 new_wire = wire[0:10] + struct.pack("!H", adcount) + wire[12:tsig_start] 

295 if rdata.error != 0: 

296 if rdata.error == dns.rcode.BADSIG: 

297 raise PeerBadSignature 

298 elif rdata.error == dns.rcode.BADKEY: 

299 raise PeerBadKey 

300 elif rdata.error == dns.rcode.BADTIME: 

301 raise PeerBadTime 

302 elif rdata.error == dns.rcode.BADTRUNC: 

303 raise PeerBadTruncation 

304 else: 

305 raise PeerError(f"unknown TSIG error code {rdata.error}") 

306 if key.name != owner: 

307 raise BadKey 

308 if key.algorithm != rdata.algorithm: 

309 raise BadAlgorithm 

310 ctx = _digest(new_wire, key, rdata, None, request_mac, ctx, multi) 

311 ctx.verify(rdata.mac) 

312 if abs(rdata.time_signed - now) > rdata.fudge: 

313 raise BadTime 

314 return _maybe_start_digest(key, rdata.mac, multi) 

315 

316 

317def get_context(key): 

318 """Returns an HMAC context for the specified key. 

319 

320 @rtype: HMAC context 

321 @raises NotImplementedError: I{algorithm} is not supported 

322 """ 

323 

324 if key.algorithm == GSS_TSIG: 

325 return GSSTSig(key.secret) 

326 else: 

327 return HMACTSig(key.secret, key.algorithm) 

328 

329 

330class Key: 

331 def __init__( 

332 self, 

333 name: dns.name.Name | str, 

334 secret: bytes | str, 

335 algorithm: dns.name.Name | str = default_algorithm, 

336 ): 

337 if isinstance(name, str): 

338 name = dns.name.from_text(name) 

339 self.name = name 

340 if isinstance(secret, str): 

341 secret = base64.decodebytes(secret.encode()) 

342 self.secret = secret 

343 if isinstance(algorithm, str): 

344 algorithm = dns.name.from_text(algorithm) 

345 self.algorithm = algorithm 

346 

347 def __eq__(self, other): 

348 return ( 

349 isinstance(other, Key) 

350 and self.name == other.name 

351 and self.secret == other.secret 

352 and self.algorithm == other.algorithm 

353 ) 

354 

355 def __repr__(self): 

356 r = f"<DNS key name='{self.name}', " + f"algorithm='{self.algorithm}'" 

357 if self.algorithm != GSS_TSIG: 

358 r += f", secret='{base64.b64encode(self.secret).decode()}'" 

359 r += ">" 

360 return r