Coverage for ivatar/utils.py: 68%

1""" 

2Simple module providing reusable random_string function 

3""" 

4 

5import contextlib 

6import http.client 

7import random 

8import string 

9import logging 

10from io import BytesIO 

11from urllib.parse import urlparse 

12from urllib.error import URLError 

13from urllib.request import urlopen as urlopen_orig 

14 

15import requests 

16from PIL import Image, ImageDraw, ImageSequence 

17from django.conf import settings 

18 

19# Initialize logger 

20logger = logging.getLogger("ivatar") 

21 

22# Access settings through django.conf.settings 

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

24URL_TIMEOUT = getattr(settings, "URL_TIMEOUT", 10) 

25BLUESKY_IDENTIFIER = getattr(settings, "BLUESKY_IDENTIFIER", None) 

26BLUESKY_APP_PASSWORD = getattr(settings, "BLUESKY_APP_PASSWORD", None) 

27 

28 

29def urlopen(url, timeout=URL_TIMEOUT): 

30 ctx = None 

31 if DEBUG: 

32 import ssl 

33 

34 ctx = ssl.create_default_context() 

35 ctx.check_hostname = False 

36 ctx.verify_mode = ssl.CERT_NONE 

37 

38 try: 

39 return urlopen_orig(url, timeout=timeout, context=ctx) 

40 except Exception as exc: 

41 # Handle malformed URLs and other HTTP client errors gracefully 

42 if isinstance(exc, http.client.InvalidURL): 

43 logger.warning( 

44 f"Invalid URL detected (possible injection attempt): {url!r} - {exc}" 

45 ) 

46 # Re-raise as URLError to maintain compatibility with existing error handling 

47 raise URLError(f"Invalid URL: {exc}") from exc 

48 elif isinstance(exc, (ValueError, UnicodeError)): 

49 logger.warning(f"Malformed URL detected: {url!r} - {exc}") 

50 raise URLError(f"Malformed URL: {exc}") from exc 

51 else: 

52 # Re-raise other exceptions as-is 

53 raise 

54 

55 

56class Bluesky: 

57 """ 

58 Handle Bluesky client access with persistent session management 

59 """ 

60 

61 identifier = "" 

62 app_password = "" 

63 service = "https://bsky.social" 

64 session = None 

65 _shared_session = None # Class-level shared session 

66 _session_expires_at = None # Track session expiration 

67 

68 def __init__( 

69 self, 

70 identifier: str = BLUESKY_IDENTIFIER, 

71 app_password: str = BLUESKY_APP_PASSWORD, 

72 service: str = "https://bsky.social", 

73 ): 

74 self.identifier = identifier 

75 self.app_password = app_password 

76 self.service = service 

77 

78 def _is_session_valid(self) -> bool: 

79 """ 

80 Check if the current session is still valid 

81 """ 

82 if not self._shared_session or not self._session_expires_at: 

83 return False 

84 

85 import time 

86 

87 # Add 5 minute buffer before actual expiration 

88 return time.time() < (self._session_expires_at - 300) 

89 

90 def login(self): 

91 """ 

92 Login to Bluesky with session persistence 

93 """ 

94 # Use shared session if available and valid 

95 if self._is_session_valid(): 

96 self.session = self._shared_session 

97 logger.debug("Reusing existing Bluesky session") 

98 return 

99 

100 logger.debug("Creating new Bluesky session") 

101 auth_response = requests.post( 

102 f"{self.service}/xrpc/com.atproto.server.createSession", 

103 json={"identifier": self.identifier, "password": self.app_password}, 

104 ) 

105 auth_response.raise_for_status() 

106 self.session = auth_response.json() 

107 

108 # Store session data for reuse 

109 self._shared_session = self.session 

110 import time 

111 

112 # Sessions typically expire in 24 hours, but we'll refresh every 12 hours 

113 self._session_expires_at = time.time() + (12 * 60 * 60) 

114 

115 logger.debug( 

116 "Created new Bluesky session, expires at: %s", 

117 time.strftime( 

118 "%Y-%m-%d %H:%M:%S", time.localtime(self._session_expires_at) 

119 ), 

120 ) 

121 

122 @classmethod 

123 def clear_shared_session(cls): 

124 """ 

125 Clear the shared session (useful for testing) 

126 """ 

127 cls._shared_session = None 

128 cls._session_expires_at = None 

129 logger.debug("Cleared shared Bluesky session") 

130 

131 def normalize_handle(self, handle: str) -> str: 

132 """ 

133 Return the normalized handle for given handle 

134 """ 

135 # Normalize Bluesky handle in case someone enters an '@' at the beginning 

136 while handle.startswith("@"): 

137 handle = handle[1:] 

138 # Remove trailing spaces or spaces at the beginning 

139 while handle.startswith(" "): 

140 handle = handle[1:] 

141 while handle.endswith(" "): 

142 handle = handle[:-1] 

143 return handle 

144 

145 def _make_profile_request(self, handle: str): 

146 """ 

147 Make a profile request to Bluesky API with automatic retry on session expiration 

148 """ 

149 try: 

150 profile_response = requests.get( 

151 f"{self.service}/xrpc/app.bsky.actor.getProfile", 

152 headers={"Authorization": f'Bearer {self.session["accessJwt"]}'}, 

153 params={"actor": handle}, 

154 ) 

155 profile_response.raise_for_status() 

156 return profile_response.json() 

157 except requests.exceptions.HTTPError as exc: 

158 if exc.response.status_code == 401: 

159 # Session expired, try to login again 

160 logger.warning("Bluesky session expired, re-authenticating") 

161 self.clear_shared_session() 

162 self.login() 

163 # Retry the request 

164 profile_response = requests.get( 

165 f"{self.service}/xrpc/app.bsky.actor.getProfile", 

166 headers={"Authorization": f'Bearer {self.session["accessJwt"]}'}, 

167 params={"actor": handle}, 

168 ) 

169 profile_response.raise_for_status() 

170 return profile_response.json() 

171 else: 

172 logger.warning(f"Bluesky profile fetch failed with HTTP error: {exc}") 

173 return None 

174 except Exception as exc: 

175 logger.warning(f"Bluesky profile fetch failed with error: {exc}") 

176 return None 

177 

178 def get_profile(self, handle: str) -> str: 

179 if not self.session or not self._is_session_valid(): 

180 self.login() 

181 return self._make_profile_request(handle) 

182 

183 def get_avatar(self, handle: str): 

184 """ 

185 Get avatar URL for a handle 

186 """ 

187 profile = self.get_profile(handle) 

188 return profile["avatar"] if profile else None 

189 

190 

191def random_string(length=10): 

192 """ 

193 Return some random string with default length 10 

194 """ 

195 return "".join( 

196 random.SystemRandom().choice(string.ascii_lowercase + string.digits) 

197 for _ in range(length) 

198 ) 

199 

200 

201def generate_random_email(): 

202 """ 

203 Generate a random email address using the same pattern as test_views.py 

204 """ 

205 username = random_string() 

206 domain = random_string() 

207 tld = random_string(2) 

208 return f"{username}@{domain}.{tld}" 

209 

210 

211def random_ip_address(): 

212 """ 

213 Return a random IP address (IPv4) 

214 """ 

215 return f"{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}" 

216 

217 

218def openid_variations(openid): 

219 """ 

220 Return the various OpenID variations, ALWAYS in the same order: 

221 - http w/ trailing slash 

222 - http w/o trailing slash 

223 - https w/ trailing slash 

224 - https w/o trailing slash 

225 """ 

226 

227 # Make the 'base' version: http w/ trailing slash 

228 if openid.startswith("https://"): 

229 openid = openid.replace("https://", "http://") 

230 if openid[-1] != "/": 

231 openid = f"{openid}/" 

232 

233 # http w/o trailing slash 

234 var1 = openid[:-1] 

235 var2 = openid.replace("http://", "https://") 

236 var3 = var2[:-1] 

237 return (openid, var1, var2, var3) 

238 

239 

240def mm_ng( 

241 idhash, size=80, add_red=0, add_green=0, add_blue=0 

242): # pylint: disable=too-many-locals 

243 """ 

244 Return an MM (mystery man) image, based on a given hash 

245 add some red, green or blue, if specified 

246 """ 

247 

248 # Make sure the lightest bg color we paint is e0, else 

249 # we do not see the MM any more 

250 if idhash[0] == "f": 

251 idhash = "e0" 

252 

253 # How large is the circle? 

254 circle_size = size * 0.6 

255 

256 # Coordinates for the circle 

257 start_x = int(size * 0.2) 

258 end_x = start_x + circle_size 

259 start_y = int(size * 0.05) 

260 end_y = start_y + circle_size 

261 

262 # All are the same, based on the input hash 

263 # this should always result in a "gray-ish" background 

264 red = idhash[:2] 

265 green = idhash[:2] 

266 blue = idhash[:2] 

267 

268 # Add some red (i/a) and make sure it's not over 255 

269 red = hex(int(red, 16) + add_red).replace("0x", "") 

270 if int(red, 16) > 255: 

271 red = "ff" 

272 if len(red) == 1: 

273 red = f"0{red}" 

274 

275 # Add some green (i/a) and make sure it's not over 255 

276 green = hex(int(green, 16) + add_green).replace("0x", "") 

277 if int(green, 16) > 255: 

278 green = "ff" 

279 if len(green) == 1: 

280 green = f"0{green}" 

281 

282 # Add some blue (i/a) and make sure it's not over 255 

283 blue = hex(int(blue, 16) + add_blue).replace("0x", "") 

284 if int(blue, 16) > 255: 

285 blue = "ff" 

286 if len(blue) == 1: 

287 blue = f"0{blue}" 

288 

289 # Assemble the bg color "string" in web notation. Eg. '#d3d3d3' 

290 bg_color = f"#{red}{green}{blue}" 

291 

292 # Image 

293 image = Image.new("RGB", (size, size)) 

294 draw = ImageDraw.Draw(image) 

295 

296 # Draw background 

297 draw.rectangle(((0, 0), (size, size)), fill=bg_color) 

298 

299 # Draw MMs head 

300 draw.ellipse((start_x, start_y, end_x, end_y), fill="white") 

301 

302 # Draw MMs 'body' 

303 draw.polygon( 

304 ( 

305 (start_x + circle_size / 2, size / 2.5), 

306 (size * 0.15, size), 

307 (size - size * 0.15, size), 

308 ), 

309 fill="white", 

310 ) 

311 

312 return image 

313 

314 

315def is_trusted_url(url, url_filters): 

316 """ 

317 Check if a URL is valid and considered a trusted URL. 

318 If the URL is malformed, returns False. 

319 

320 Based on: https://developer.mozilla.org/en-US/docs/Mozilla/Add-ons/WebExtensions/API/events/UrlFilter 

321 """ 

322 scheme, netloc, path, params, query, fragment = urlparse(url) 

323 

324 for ufilter in url_filters: 

325 if "schemes" in ufilter: 

326 schemes = ufilter["schemes"] 

327 

328 if scheme not in schemes: 

329 continue 

330 

331 if "host_equals" in ufilter: 

332 host_equals = ufilter["host_equals"] 

333 

334 if netloc != host_equals: 

335 continue 

336 

337 if "host_suffix" in ufilter: 

338 host_suffix = ufilter["host_suffix"] 

339 

340 if not netloc.endswith(host_suffix): 

341 continue 

342 

343 if "path_prefix" in ufilter: 

344 path_prefix = ufilter["path_prefix"] 

345 

346 if not path.startswith(path_prefix): 

347 continue 

348 

349 if "url_prefix" in ufilter: 

350 url_prefix = ufilter["url_prefix"] 

351 

352 if not url.startswith(url_prefix): 

353 continue 

354 

355 return True 

356 

357 return False 

358 

359 

360def resize_animated_gif(input_pil: Image, size: list) -> BytesIO: 

361 def _thumbnail_frames(image): 

362 for frame in ImageSequence.Iterator(image): 

363 new_frame = frame.copy() 

364 new_frame.thumbnail(size) 

365 yield new_frame 

366 

367 frames = list(_thumbnail_frames(input_pil)) 

368 output = BytesIO() 

369 output_image = frames[0] 

370 output_image.save( 

371 output, 

372 format="gif", 

373 save_all=True, 

374 optimize=False, 

375 append_images=frames[1:], 

376 disposal=input_pil.disposal_method, 

377 **input_pil.info, 

378 ) 

379 return output