Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/aiodns/compat.py: 59%

1""" 

2Compatibility layer for pycares 5.x API. 

3 

4This module provides result types compatible with pycares 4.x API 

5to maintain backward compatibility with existing code. 

6""" 

7 

8from __future__ import annotations 

9 

10from dataclasses import dataclass 

11from typing import Union, cast 

12 

13import pycares 

14 

15from . import error 

16 

17_SINGLE_RESULT_QTYPES = frozenset( 

18 { 

19 pycares.QUERY_TYPE_CNAME, 

20 pycares.QUERY_TYPE_SOA, 

21 pycares.QUERY_TYPE_PTR, 

22 } 

23) 

24 

25 

26def _maybe_str(data: bytes) -> str | bytes: 

27 """Decode bytes as ASCII, return bytes if decode fails (pycares 4.x).""" 

28 try: 

29 return data.decode('ascii') 

30 except UnicodeDecodeError: 

31 return data 

32 

33 

34@dataclass(frozen=True, slots=True) 

35class AresQueryAResult: 

36 """A record result (compatible with pycares 4.x ares_query_a_result).""" 

37 

38 host: str 

39 ttl: int 

40 

41 

42@dataclass(frozen=True, slots=True) 

43class AresQueryAAAAResult: 

44 """AAAA record result (pycares 4.x compat).""" 

45 

46 host: str 

47 ttl: int 

48 

49 

50@dataclass(frozen=True, slots=True) 

51class AresQueryCNAMEResult: 

52 """CNAME record result (pycares 4.x compat).""" 

53 

54 cname: str 

55 ttl: int 

56 

57 

58@dataclass(frozen=True, slots=True) 

59class AresQueryMXResult: 

60 """MX record result (pycares 4.x compat).""" 

61 

62 host: str 

63 priority: int 

64 ttl: int 

65 

66 

67@dataclass(frozen=True, slots=True) 

68class AresQueryNSResult: 

69 """NS record result (pycares 4.x compat).""" 

70 

71 host: str 

72 ttl: int 

73 

74 

75@dataclass(frozen=True, slots=True) 

76class AresQueryTXTResult: 

77 """TXT record result (pycares 4.x compat).""" 

78 

79 text: str | bytes # str if ASCII, bytes otherwise (pycares 4.x behavior) 

80 ttl: int 

81 

82 

83@dataclass(frozen=True, slots=True) 

84class AresQuerySOAResult: 

85 """SOA record result (pycares 4.x compat).""" 

86 

87 nsname: str 

88 hostmaster: str 

89 serial: int 

90 refresh: int 

91 retry: int 

92 expires: int 

93 minttl: int 

94 ttl: int 

95 

96 

97@dataclass(frozen=True, slots=True) 

98class AresQuerySRVResult: 

99 """SRV record result (pycares 4.x compat).""" 

100 

101 host: str 

102 port: int 

103 priority: int 

104 weight: int 

105 ttl: int 

106 

107 

108@dataclass(frozen=True, slots=True) 

109class AresQueryNAPTRResult: 

110 """NAPTR record result (pycares 4.x compat).""" 

111 

112 order: int 

113 preference: int 

114 flags: str 

115 service: str 

116 regex: str 

117 replacement: str 

118 ttl: int 

119 

120 

121@dataclass(frozen=True, slots=True) 

122class AresQueryCAAResult: 

123 """CAA record result (pycares 4.x compat).""" 

124 

125 critical: int 

126 property: str 

127 value: str 

128 ttl: int 

129 

130 

131@dataclass(frozen=True, slots=True) 

132class AresQueryPTRResult: 

133 """PTR record result (pycares 4.x compat).""" 

134 

135 name: str 

136 ttl: int 

137 aliases: list[str] 

138 

139 

140@dataclass(frozen=True, slots=True) 

141class AresHostResult: 

142 """Host result (compatible with pycares 4.x ares_host_result).""" 

143 

144 name: str 

145 aliases: list[str] 

146 addresses: list[str] 

147 

148 

149# Type alias for a single converted record 

150ConvertedRecord = Union[ 

151 AresQueryAResult, 

152 AresQueryAAAAResult, 

153 AresQueryCNAMEResult, 

154 AresQueryMXResult, 

155 AresQueryNSResult, 

156 AresQueryTXTResult, 

157 AresQuerySOAResult, 

158 AresQuerySRVResult, 

159 AresQueryNAPTRResult, 

160 AresQueryCAAResult, 

161 AresQueryPTRResult, 

162 pycares.DNSRecord, # Unknown types returned as-is 

163] 

164 

165# Type alias for query results 

166QueryResult = Union[ 

167 list[AresQueryAResult], 

168 list[AresQueryAAAAResult], 

169 AresQueryCNAMEResult, 

170 list[AresQueryMXResult], 

171 list[AresQueryNSResult], 

172 list[AresQueryTXTResult], 

173 AresQuerySOAResult, 

174 list[AresQuerySRVResult], 

175 list[AresQueryNAPTRResult], 

176 list[AresQueryCAAResult], 

177 AresQueryPTRResult, 

178 list[ConvertedRecord], # For ANY query type 

179] 

180 

181 

182def _convert_record(record: pycares.DNSRecord) -> ConvertedRecord: 

183 """Convert a single DNS record to pycares 4.x compatible format.""" 

184 ttl = record.ttl 

185 record_type = record.type 

186 

187 if record_type == pycares.QUERY_TYPE_A: 

188 a_data = cast(pycares.ARecordData, record.data) 

189 return AresQueryAResult(host=a_data.addr, ttl=ttl) 

190 if record_type == pycares.QUERY_TYPE_AAAA: 

191 aaaa_data = cast(pycares.AAAARecordData, record.data) 

192 return AresQueryAAAAResult(host=aaaa_data.addr, ttl=ttl) 

193 if record_type == pycares.QUERY_TYPE_CNAME: 

194 cname_data = cast(pycares.CNAMERecordData, record.data) 

195 return AresQueryCNAMEResult(cname=cname_data.cname, ttl=ttl) 

196 if record_type == pycares.QUERY_TYPE_MX: 

197 mx_data = cast(pycares.MXRecordData, record.data) 

198 return AresQueryMXResult( 

199 host=mx_data.exchange, priority=mx_data.priority, ttl=ttl 

200 ) 

201 if record_type == pycares.QUERY_TYPE_NS: 

202 ns_data = cast(pycares.NSRecordData, record.data) 

203 return AresQueryNSResult(host=ns_data.nsdname, ttl=ttl) 

204 if record_type == pycares.QUERY_TYPE_TXT: 

205 txt_data = cast(pycares.TXTRecordData, record.data) 

206 return AresQueryTXTResult(text=_maybe_str(txt_data.data), ttl=ttl) 

207 if record_type == pycares.QUERY_TYPE_SOA: 

208 soa_data = cast(pycares.SOARecordData, record.data) 

209 return AresQuerySOAResult( 

210 nsname=soa_data.mname, 

211 hostmaster=soa_data.rname, 

212 serial=soa_data.serial, 

213 refresh=soa_data.refresh, 

214 retry=soa_data.retry, 

215 expires=soa_data.expire, 

216 minttl=soa_data.minimum, 

217 ttl=ttl, 

218 ) 

219 if record_type == pycares.QUERY_TYPE_SRV: 

220 srv_data = cast(pycares.SRVRecordData, record.data) 

221 return AresQuerySRVResult( 

222 host=srv_data.target, 

223 port=srv_data.port, 

224 priority=srv_data.priority, 

225 weight=srv_data.weight, 

226 ttl=ttl, 

227 ) 

228 if record_type == pycares.QUERY_TYPE_NAPTR: 

229 naptr_data = cast(pycares.NAPTRRecordData, record.data) 

230 return AresQueryNAPTRResult( 

231 order=naptr_data.order, 

232 preference=naptr_data.preference, 

233 flags=naptr_data.flags, 

234 service=naptr_data.service, 

235 regex=naptr_data.regexp, 

236 replacement=naptr_data.replacement, 

237 ttl=ttl, 

238 ) 

239 if record_type == pycares.QUERY_TYPE_CAA: 

240 caa_data = cast(pycares.CAARecordData, record.data) 

241 return AresQueryCAAResult( 

242 critical=caa_data.critical, 

243 property=caa_data.tag, 

244 value=caa_data.value, 

245 ttl=ttl, 

246 ) 

247 if record_type == pycares.QUERY_TYPE_PTR: 

248 ptr_data = cast(pycares.PTRRecordData, record.data) 

249 return AresQueryPTRResult(name=ptr_data.dname, ttl=ttl, aliases=[]) 

250 # Return raw record for unknown types 

251 return record 

252 

253 

254def convert_result(dns_result: pycares.DNSResult, qtype: int) -> QueryResult: 

255 """Convert pycares 5.x DNSResult to pycares 4.x compatible format.""" 

256 # For ANY - convert all records and return mixed list 

257 if qtype == pycares.QUERY_TYPE_ANY: 

258 return [_convert_record(record) for record in dns_result.answer] 

259 

260 results: list[ConvertedRecord] = [] 

261 

262 for record in dns_result.answer: 

263 record_type = record.type 

264 

265 # Filter by query type since answer can contain other types 

266 # (e.g., CNAME records when querying for A/AAAA) 

267 if record_type != qtype: 

268 continue 

269 

270 converted = _convert_record(record) 

271 

272 # CNAME, SOA, and PTR return single result, not list 

273 if record_type in _SINGLE_RESULT_QTYPES: 

274 return cast(QueryResult, converted) 

275 

276 results.append(converted) 

277 

278 # NOERROR/NODATA: c-ares delivered ARES_SUCCESS but the answer has no 

279 # records of the queried type. pycares 4.x raised ARES_ENODATA here; 

280 # without this branch single-result qtypes (CNAME/SOA/PTR) would 

281 # resolve to [] and crash callers reading .name/.cname/.nsname. 

282 if not results: 

283 raise error.DNSError( 

284 pycares.errno.ARES_ENODATA, 

285 pycares.errno.strerror(pycares.errno.ARES_ENODATA), 

286 ) 

287 

288 return results