Coverage for ivatar/utils.py: 70%
201 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-20 05:10 +0000
1"""
2Simple module providing reusable random_string function
3"""
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
15import requests
16from PIL import Image, ImageDraw, ImageSequence
17from django.conf import settings
19# Initialize logger
20logger = logging.getLogger("ivatar")
22# Access settings through django.conf.settings
23DEBUG = getattr(settings, "DEBUG", False)
24URL_TIMEOUT = getattr(settings, "URL_TIMEOUT", 10)
25BLUESKY_API_TIMEOUT = getattr(settings, "BLUESKY_API_TIMEOUT", 5)
26BLUESKY_IDENTIFIER = getattr(settings, "BLUESKY_IDENTIFIER", None)
27BLUESKY_APP_PASSWORD = getattr(settings, "BLUESKY_APP_PASSWORD", None)
30def urlopen(url, timeout=URL_TIMEOUT):
31 ctx = None
32 if DEBUG:
33 import ssl
35 ctx = ssl.create_default_context()
36 ctx.check_hostname = False
37 ctx.verify_mode = ssl.CERT_NONE
39 try:
40 return urlopen_orig(url, timeout=timeout, context=ctx)
41 except Exception as exc:
42 # Handle malformed URLs and other HTTP client errors gracefully
43 if isinstance(exc, http.client.InvalidURL):
44 logger.warning(
45 f"Invalid URL detected (possible injection attempt): {url!r} - {exc}"
46 )
47 # Re-raise as URLError to maintain compatibility with existing error handling
48 raise URLError(f"Invalid URL: {exc}") from exc
49 elif isinstance(exc, (ValueError, UnicodeError)):
50 logger.warning(f"Malformed URL detected: {url!r} - {exc}")
51 raise URLError(f"Malformed URL: {exc}") from exc
52 else:
53 # Re-raise other exceptions as-is
54 raise
57class Bluesky:
58 """
59 Handle Bluesky client access with persistent session management
60 """
62 identifier = ""
63 app_password = ""
64 service = "https://bsky.social"
65 session = None
66 _shared_session = None # Class-level shared session
67 _session_expires_at = None # Track session expiration
69 def __init__(
70 self,
71 identifier: str = BLUESKY_IDENTIFIER,
72 app_password: str = BLUESKY_APP_PASSWORD,
73 service: str = "https://bsky.social",
74 ):
75 self.identifier = identifier
76 self.app_password = app_password
77 self.service = service
79 def _is_session_valid(self) -> bool:
80 """
81 Check if the current session is still valid
82 """
83 if not self._shared_session or not self._session_expires_at:
84 return False
86 import time
88 # Add 5 minute buffer before actual expiration
89 return time.time() < (self._session_expires_at - 300)
91 def login(self):
92 """
93 Login to Bluesky with session persistence
94 """
95 # Use shared session if available and valid
96 if self._is_session_valid():
97 self.session = self._shared_session
98 logger.debug("Reusing existing Bluesky session")
99 return
101 logger.debug("Creating new Bluesky session")
102 auth_response = requests.post(
103 f"{self.service}/xrpc/com.atproto.server.createSession",
104 json={"identifier": self.identifier, "password": self.app_password},
105 timeout=BLUESKY_API_TIMEOUT,
106 )
107 auth_response.raise_for_status()
108 self.session = auth_response.json()
110 # Store session data for reuse
111 self._shared_session = self.session
112 import time
114 # Sessions typically expire in 24 hours, but we'll refresh every 12 hours
115 self._session_expires_at = time.time() + (12 * 60 * 60)
117 logger.debug(
118 "Created new Bluesky session, expires at: %s",
119 time.strftime(
120 "%Y-%m-%d %H:%M:%S", time.localtime(self._session_expires_at)
121 ),
122 )
124 @classmethod
125 def clear_shared_session(cls):
126 """
127 Clear the shared session (useful for testing)
128 """
129 cls._shared_session = None
130 cls._session_expires_at = None
131 logger.debug("Cleared shared Bluesky session")
133 def normalize_handle(self, handle: str) -> str:
134 """
135 Return the normalized handle for given handle
136 """
137 # Normalize Bluesky handle in case someone enters an '@' at the beginning
138 while handle.startswith("@"):
139 handle = handle[1:]
140 # Remove trailing spaces or spaces at the beginning
141 while handle.startswith(" "):
142 handle = handle[1:]
143 while handle.endswith(" "):
144 handle = handle[:-1]
145 return handle
147 def _make_profile_request(self, handle: str):
148 """
149 Make a profile request to Bluesky API with automatic retry on session expiration
150 """
151 try:
152 profile_response = requests.get(
153 f"{self.service}/xrpc/app.bsky.actor.getProfile",
154 headers={"Authorization": f'Bearer {self.session["accessJwt"]}'},
155 params={"actor": handle},
156 timeout=BLUESKY_API_TIMEOUT,
157 )
158 profile_response.raise_for_status()
159 return profile_response.json()
160 except requests.exceptions.HTTPError as exc:
161 if exc.response.status_code == 401:
162 # Session expired, try to login again
163 logger.warning("Bluesky session expired, re-authenticating")
164 self.clear_shared_session()
165 self.login()
166 # Retry the request
167 profile_response = requests.get(
168 f"{self.service}/xrpc/app.bsky.actor.getProfile",
169 headers={"Authorization": f'Bearer {self.session["accessJwt"]}'},
170 params={"actor": handle},
171 timeout=BLUESKY_API_TIMEOUT,
172 )
173 profile_response.raise_for_status()
174 return profile_response.json()
175 else:
176 logger.warning(f"Bluesky profile fetch failed with HTTP error: {exc}")
177 return None
178 except Exception as exc:
179 logger.warning(f"Bluesky profile fetch failed with error: {exc}")
180 return None
182 def get_profile(self, handle: str) -> str:
183 if not self.session or not self._is_session_valid():
184 self.login()
185 return self._make_profile_request(handle)
187 def get_avatar(self, handle: str):
188 """
189 Get avatar URL for a handle
190 """
191 profile = self.get_profile(handle)
192 return profile["avatar"] if profile else None
195def random_string(length=10):
196 """
197 Return some random string with default length 10
198 """
199 return "".join(
200 random.SystemRandom().choice(string.ascii_lowercase + string.digits)
201 for _ in range(length)
202 )
205def generate_random_email():
206 """
207 Generate a random email address using the same pattern as test_views.py
208 """
209 username = random_string()
210 domain = random_string()
211 tld = random_string(2)
212 return f"{username}@{domain}.{tld}"
215def random_ip_address():
216 """
217 Return a random IP address (IPv4)
218 """
219 return f"{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}.{random.randint(1, 254)}"
222def openid_variations(openid):
223 """
224 Return the various OpenID variations, ALWAYS in the same order:
225 - http w/ trailing slash
226 - http w/o trailing slash
227 - https w/ trailing slash
228 - https w/o trailing slash
229 """
231 # Make the 'base' version: http w/ trailing slash
232 if openid.startswith("https://"):
233 openid = openid.replace("https://", "http://")
234 if openid[-1] != "/":
235 openid = f"{openid}/"
237 # http w/o trailing slash
238 var1 = openid[:-1]
239 var2 = openid.replace("http://", "https://")
240 var3 = var2[:-1]
241 return (openid, var1, var2, var3)
244def mm_ng(
245 idhash, size=80, add_red=0, add_green=0, add_blue=0
246): # pylint: disable=too-many-locals
247 """
248 Return an MM (mystery man) image, based on a given hash
249 add some red, green or blue, if specified
250 """
252 # Make sure the lightest bg color we paint is e0, else
253 # we do not see the MM any more
254 if idhash[0] == "f":
255 idhash = "e0"
257 # How large is the circle?
258 circle_size = size * 0.6
260 # Coordinates for the circle
261 start_x = int(size * 0.2)
262 end_x = start_x + circle_size
263 start_y = int(size * 0.05)
264 end_y = start_y + circle_size
266 # All are the same, based on the input hash
267 # this should always result in a "gray-ish" background
268 red = idhash[:2]
269 green = idhash[:2]
270 blue = idhash[:2]
272 # Add some red (i/a) and make sure it's not over 255
273 red = hex(int(red, 16) + add_red).replace("0x", "")
274 if int(red, 16) > 255:
275 red = "ff"
276 if len(red) == 1:
277 red = f"0{red}"
279 # Add some green (i/a) and make sure it's not over 255
280 green = hex(int(green, 16) + add_green).replace("0x", "")
281 if int(green, 16) > 255:
282 green = "ff"
283 if len(green) == 1:
284 green = f"0{green}"
286 # Add some blue (i/a) and make sure it's not over 255
287 blue = hex(int(blue, 16) + add_blue).replace("0x", "")
288 if int(blue, 16) > 255:
289 blue = "ff"
290 if len(blue) == 1:
291 blue = f"0{blue}"
293 # Assemble the bg color "string" in web notation. Eg. '#d3d3d3'
294 bg_color = f"#{red}{green}{blue}"
296 # Image
297 image = Image.new("RGB", (size, size))
298 draw = ImageDraw.Draw(image)
300 # Draw background
301 draw.rectangle(((0, 0), (size, size)), fill=bg_color)
303 # Draw MMs head
304 draw.ellipse((start_x, start_y, end_x, end_y), fill="white")
306 # Draw MMs 'body'
307 draw.polygon(
308 (
309 (start_x + circle_size / 2, size / 2.5),
310 (size * 0.15, size),
311 (size - size * 0.15, size),
312 ),
313 fill="white",
314 )
316 return image
319def is_trusted_url(url, url_filters):
320 """
321 Check if a URL is valid and considered a trusted URL.
322 If the URL is malformed, returns False.
324 Based on: https://developer.mozilla.org/en-US/docs/Mozilla/Add-ons/WebExtensions/API/events/UrlFilter
325 """
326 scheme, netloc, path, params, query, fragment = urlparse(url)
328 for ufilter in url_filters:
329 if "schemes" in ufilter:
330 schemes = ufilter["schemes"]
332 if scheme not in schemes:
333 continue
335 if "host_equals" in ufilter:
336 host_equals = ufilter["host_equals"]
338 if netloc != host_equals:
339 continue
341 if "host_suffix" in ufilter:
342 host_suffix = ufilter["host_suffix"]
344 if not netloc.endswith(host_suffix):
345 continue
347 if "path_prefix" in ufilter:
348 path_prefix = ufilter["path_prefix"]
350 if not path.startswith(path_prefix):
351 continue
353 if "url_prefix" in ufilter:
354 url_prefix = ufilter["url_prefix"]
356 if not url.startswith(url_prefix):
357 continue
359 return True
361 return False
364def resize_animated_gif(input_pil: Image, size: list) -> BytesIO:
365 def _thumbnail_frames(image):
366 for frame in ImageSequence.Iterator(image):
367 new_frame = frame.copy()
368 new_frame.thumbnail(size)
369 yield new_frame
371 frames = list(_thumbnail_frames(input_pil))
372 output = BytesIO()
373 output_image = frames[0]
374 output_image.save(
375 output,
376 format="gif",
377 save_all=True,
378 optimize=False,
379 append_images=frames[1:],
380 disposal=input_pil.disposal_method,
381 **input_pil.info,
382 )
383 return output
386def get_resample_filter(target_size: int, original_size=None):
387 """
388 Select the most efficient resampling filter based on target size and scaling direction.
389 - For small downscaled targets (<= 64px): Image.BOX (box-averaging, 3.8x faster than Lanczos)
390 - For medium targets (<= 128px): Image.BILINEAR (fast linear interpolation, 1.4x faster)
391 - For large targets or upscaling small sources: Image.LANCZOS (high-quality sinc filter)
392 """
393 if original_size is not None:
394 if isinstance(original_size, (tuple, list)):
395 orig_dim = min(original_size)
396 else:
397 orig_dim = original_size
398 # If upscaling from a smaller image to larger size, use Lanczos for quality
399 if orig_dim < target_size:
400 return Image.LANCZOS
402 if target_size <= 64:
403 return Image.BOX
404 if target_size <= 128:
405 return Image.BILINEAR
406 return Image.LANCZOS