Coverage for ivatar/tools/views.py: 67%

1""" 

2View classes for ivatar/tools/ 

3""" 

4 

5import hashlib 

6import random 

7 

8from django.views.generic.edit import FormView 

9from django.urls import reverse_lazy as reverse 

10from django.shortcuts import render 

11 

12import dns.resolver 

13import dns.exception 

14 

15from libravatar import libravatar_url, parse_user_identity 

16from libravatar import SECURE_BASE_URL as LIBRAVATAR_SECURE_BASE_URL 

17from libravatar import BASE_URL as LIBRAVATAR_BASE_URL 

18 

19from django.conf import settings 

20 

21# Access settings through django.conf.settings 

22SECURE_BASE_URL = getattr(settings, "SECURE_BASE_URL", "https://avatars.linux-kernel.at/avatar/") 

23BASE_URL = getattr(settings, "BASE_URL", "http://avatars.linux-kernel.at/avatar/") 

24SITE_NAME = getattr(settings, "SITE_NAME", "libravatar") 

25DEBUG = getattr(settings, "DEBUG", False) 

26 

27from .forms import ( 

28 CheckDomainForm, 

29 CheckForm, 

30) # pylint: disable=relative-beyond-top-level 

31 

32 

33class CheckDomainView(FormView): 

34 """ 

35 View class for checking a domain 

36 """ 

37 

38 template_name = "check_domain.html" 

39 form_class = CheckDomainForm 

40 success_url = reverse("tools_check_domain") 

41 

42 def form_valid(self, form): 

43 super().form_valid(form) 

44 domain = form.cleaned_data["domain"] 

45 result = {"avatar_server_http": lookup_avatar_server(domain, False)} 

46 if result["avatar_server_http"]: 

47 result["avatar_server_http_ipv4"] = lookup_ip_address( 

48 result["avatar_server_http"], False 

49 ) 

50 result["avatar_server_http_ipv6"] = lookup_ip_address( 

51 result["avatar_server_http"], True 

52 ) 

53 result["avatar_server_https"] = lookup_avatar_server(domain, True) 

54 if result["avatar_server_https"]: 

55 result["avatar_server_https_ipv4"] = lookup_ip_address( 

56 result["avatar_server_https"], False 

57 ) 

58 result["avatar_server_https_ipv6"] = lookup_ip_address( 

59 result["avatar_server_https"], True 

60 ) 

61 return render( 

62 self.request, 

63 self.template_name, 

64 { 

65 "form": form, 

66 "result": result, 

67 }, 

68 ) 

69 

70 

71class CheckView(FormView): 

72 """ 

73 View class for checking an e-mail or openid address 

74 """ 

75 

76 template_name = "check.html" 

77 form_class = CheckForm 

78 success_url = reverse("tools_check") 

79 

80 def form_valid(self, form): 

81 mailurl = None 

82 openidurl = None 

83 mailurl_secure = None 

84 mailurl_secure_256 = None 

85 openidurl_secure = None 

86 mail_hash = None 

87 mail_hash256 = None 

88 openid_hash = None 

89 super().form_valid(form) 

90 

91 if form.cleaned_data["default_url"]: 

92 default_url = form.cleaned_data["default_url"] 

93 elif ( 

94 form.cleaned_data["default_opt"] 

95 and form.cleaned_data["default_opt"] != "none" 

96 ): 

97 default_url = form.cleaned_data["default_opt"] 

98 else: 

99 default_url = None 

100 

101 size = form.cleaned_data["size"] if "size" in form.cleaned_data else 80 

102 if form.cleaned_data["mail"]: 

103 mailurl = libravatar_url( 

104 email=form.cleaned_data["mail"], size=size, default=default_url 

105 ) 

106 mailurl = mailurl.replace(LIBRAVATAR_BASE_URL, BASE_URL) 

107 mailurl_secure = libravatar_url( 

108 email=form.cleaned_data["mail"], 

109 size=size, 

110 https=True, 

111 default=default_url, 

112 ) 

113 mailurl_secure = mailurl_secure.replace( 

114 LIBRAVATAR_SECURE_BASE_URL, SECURE_BASE_URL 

115 ) 

116 mail_hash = parse_user_identity( 

117 email=form.cleaned_data["mail"], openid=None 

118 )[0] 

119 hash_obj = hashlib.new("sha256") 

120 hash_obj.update(form.cleaned_data["mail"].encode("utf-8")) 

121 mail_hash256 = hash_obj.hexdigest() 

122 mailurl_secure_256 = mailurl_secure.replace(mail_hash, mail_hash256) 

123 if form.cleaned_data["openid"]: 

124 if not form.cleaned_data["openid"].startswith( 

125 "http://" 

126 ) and not form.cleaned_data["openid"].startswith("https://"): 

127 form.cleaned_data["openid"] = f'http://{form.cleaned_data["openid"]}' 

128 openidurl = libravatar_url( 

129 openid=form.cleaned_data["openid"], size=size, default=default_url 

130 ) 

131 openidurl = openidurl.replace(LIBRAVATAR_BASE_URL, BASE_URL) 

132 openidurl_secure = libravatar_url( 

133 openid=form.cleaned_data["openid"], 

134 size=size, 

135 https=True, 

136 default=default_url, 

137 ) 

138 openidurl_secure = openidurl_secure.replace( 

139 LIBRAVATAR_SECURE_BASE_URL, SECURE_BASE_URL 

140 ) 

141 openid_hash = parse_user_identity( 

142 openid=form.cleaned_data["openid"], email=None 

143 )[0] 

144 

145 if "DEVELOPMENT" in SITE_NAME and DEBUG: 

146 if mailurl: 

147 mailurl = mailurl.replace( 

148 "https://avatars.linux-kernel.at", 

149 f"http://{self.request.get_host()}", 

150 ) 

151 if mailurl_secure: 

152 mailurl_secure = mailurl_secure.replace( 

153 "https://avatars.linux-kernel.at", 

154 f"http://{self.request.get_host()}", 

155 ) 

156 if mailurl_secure_256: 

157 mailurl_secure_256 = mailurl_secure_256.replace( 

158 "https://avatars.linux-kernel.at", 

159 f"http://{self.request.get_host()}", 

160 ) 

161 

162 if openidurl: 

163 openidurl = openidurl.replace( 

164 "https://avatars.linux-kernel.at", 

165 f"http://{self.request.get_host()}", 

166 ) 

167 if openidurl_secure: 

168 openidurl_secure = openidurl_secure.replace( 

169 "https://avatars.linux-kernel.at", 

170 f"http://{self.request.get_host()}", 

171 ) 

172 print(mailurl, openidurl, mailurl_secure, mailurl_secure_256, openidurl_secure) 

173 

174 return render( 

175 self.request, 

176 self.template_name, 

177 { 

178 "form": form, 

179 "mailurl": mailurl, 

180 "openidurl": openidurl, 

181 "mailurl_secure": mailurl_secure, 

182 "mailurl_secure_256": mailurl_secure_256, 

183 "openidurl_secure": openidurl_secure, 

184 "mail_hash": mail_hash, 

185 "mail_hash256": mail_hash256, 

186 "openid_hash": openid_hash, 

187 "size": size, 

188 }, 

189 ) 

190 

191 

192def lookup_avatar_server(domain, https): 

193 """ 

194 Extract the avatar server from an SRV record in the DNS zone 

195 

196 The SRV records should look like this: 

197 

198 _avatars._tcp.example.com. IN SRV 0 0 80 avatars.example.com 

199 _avatars-sec._tcp.example.com. IN SRV 0 0 443 avatars.example.com 

200 """ 

201 

202 if domain and len(domain) > 60: 

203 domain = domain[:60] 

204 

205 service_name = None 

206 if https: 

207 service_name = f"_avatars-sec._tcp.{domain}" 

208 else: 

209 service_name = f"_avatars._tcp.{domain}" 

210 

211 try: 

212 answers = dns.resolver.resolve(service_name, "SRV") 

213 except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): 

214 return None 

215 except dns.exception.DNSException as message: 

216 print(f"DNS Error: {message} ({domain})") 

217 return None 

218 

219 records = [] 

220 for rdata in answers: 

221 record = { 

222 "priority": rdata.priority, 

223 "weight": rdata.weight, 

224 "port": rdata.port, 

225 "target": rdata.target.to_text(omit_final_dot=True), 

226 } 

227 

228 records.append(record) 

229 

230 target, port = srv_hostname(records) 

231 

232 if target and ((https and port != 443) or (not https and port != 80)): 

233 return f"{target}:{port}" 

234 

235 return target 

236 

237 

238def srv_hostname(records): 

239 """ 

240 Return the right (target, port) pair from a list of SRV records. 

241 """ 

242 

243 if len(records) < 1: 

244 return (None, None) 

245 

246 if len(records) == 1: 

247 ret = records[0] 

248 return (ret["target"], ret["port"]) 

249 

250 # Keep only the servers in the top priority 

251 priority_records = [] 

252 total_weight = 0 

253 top_priority = records[0]["priority"] # highest priority = lowest number 

254 

255 for ret in records: 

256 if ret["priority"] > top_priority: 

257 # ignore the record (ret has lower priority) 

258 continue 

259 

260 # Take care - this if is only a if, if the above if 

261 # uses continue at the end. else it should be an elsif 

262 if ret["priority"] < top_priority: 

263 # reset the priority (ret has higher priority) 

264 top_priority = ret["priority"] 

265 total_weight = 0 

266 priority_records = [] 

267 

268 total_weight += ret["weight"] 

269 

270 if ret["weight"] > 0: 

271 priority_records.append((total_weight, ret)) 

272 else: 

273 # zero-weight elements must come first 

274 priority_records.insert(0, (0, ret)) 

275 

276 if len(priority_records) == 1: 

277 unused, ret = priority_records[0] # pylint: disable=unused-variable 

278 return (ret["target"], ret["port"]) 

279 

280 # Select first record according to RFC2782 weight ordering algorithm (page 3) 

281 random_number = random.randint(0, total_weight) 

282 

283 for record in priority_records: 

284 weighted_index, ret = record 

285 

286 if weighted_index >= random_number: 

287 return (ret["target"], ret["port"]) 

288 

289 print("There is something wrong with our SRV weight ordering algorithm") 

290 return (None, None) 

291 

292 

293def lookup_ip_address(hostname, ipv6): 

294 """ 

295 Try to get IPv4 or IPv6 addresses for the given hostname 

296 """ 

297 

298 try: 

299 qtype = "AAAA" if ipv6 else "A" 

300 answers = dns.resolver.resolve(hostname, qtype) 

301 for rdata in answers: 

302 return rdata.address 

303 except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): 

304 return None 

305 except dns.exception.DNSException as message: 

306 print(f"DNS Error: {message} ({hostname})") 

307 return None 

308 return None