From 7559ddad6e23b0d7a5cd162b3395ebc1d17e0f91 Mon Sep 17 00:00:00 2001 From: Oliver Falk Date: Sat, 25 Jan 2025 13:16:04 +0100 Subject: [PATCH] Refactor: Centralize URL handling and add Bluesky integration - Centralize urlopen functionality in utils.py with improved error handling - Add configurable URL timeout across the project - Introduce Bluesky client for fetching avatar from there (TBC) - Support SSL verification bypass in debug mode Signed-off-by: Oliver Falk --- config.py | 2 + ivatar/ivataraccount/gravatar.py | 7 ++-- ivatar/ivataraccount/models.py | 2 +- ivatar/ivataraccount/views.py | 2 +- ivatar/utils.py | 69 ++++++++++++++++++++++++++++++++ ivatar/views.py | 8 ++-- 6 files changed, 79 insertions(+), 11 deletions(-) diff --git a/config.py b/config.py index cb4e18c..b269786 100644 --- a/config.py +++ b/config.py @@ -286,3 +286,5 @@ def map_legacy_config(trusted_url): # Backward compability for legacy behavior TRUSTED_DEFAULT_URLS = list(map(map_legacy_config, TRUSTED_DEFAULT_URLS)) + +URL_TIMEOUT = 10 diff --git a/ivatar/ivataraccount/gravatar.py b/ivatar/ivataraccount/gravatar.py index 18ca9a7..8ee51a0 100644 --- a/ivatar/ivataraccount/gravatar.py +++ b/ivatar/ivataraccount/gravatar.py @@ -3,13 +3,12 @@ Helper method to fetch Gravatar image """ from ssl import SSLError -from urllib.request import urlopen, HTTPError, URLError +from urllib.request import HTTPError, URLError +from ivatar.utils import urlopen import hashlib from ..settings import AVATAR_MAX_SIZE -URL_TIMEOUT = 5 # in seconds - def get_photo(email): """ @@ -30,7 +29,7 @@ def get_photo(email): service_url = "http://www.gravatar.com/" + hash_object.hexdigest() try: - urlopen(image_url, timeout=URL_TIMEOUT) + urlopen(image_url) except HTTPError as exc: if exc.code != 404 and exc.code != 503: print( # pragma: no cover diff --git a/ivatar/ivataraccount/models.py b/ivatar/ivataraccount/models.py index f6217aa..b5e3472 100644 --- a/ivatar/ivataraccount/models.py +++ b/ivatar/ivataraccount/models.py @@ -9,7 +9,7 @@ import time from io import BytesIO from os import urandom from urllib.error import HTTPError, URLError -from urllib.request import urlopen +from ivatar.utils import urlopen from urllib.parse import urlsplit, urlunsplit from PIL import Image diff --git a/ivatar/ivataraccount/views.py b/ivatar/ivataraccount/views.py index f993cb5..cf370f2 100644 --- a/ivatar/ivataraccount/views.py +++ b/ivatar/ivataraccount/views.py @@ -3,7 +3,7 @@ View classes for ivatar/ivataraccount/ """ from io import BytesIO -from urllib.request import urlopen +from ivatar.utils import urlopen import base64 import binascii from xml.sax import saxutils diff --git a/ivatar/utils.py b/ivatar/utils.py index cf4a524..8336a66 100644 --- a/ivatar/utils.py +++ b/ivatar/utils.py @@ -7,6 +7,75 @@ import string from io import BytesIO from PIL import Image, ImageDraw, ImageSequence from urllib.parse import urlparse +import requests +from ivatar.settings import DEBUG, URL_TIMEOUT, BLUESKY_IDENTIFIER, BLUESKY_APP_PASSWORD +from urllib.request import urlopen as urlopen_orig + + +def urlopen(url, timeout=URL_TIMEOUT): + ctx = None + if DEBUG: + import ssl + + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE + return urlopen_orig(url, timeout=timeout, context=ctx) + + +class Bluesky: + """ + Handle Bluesky client access + """ + + identifier = "" + app_password = "" + service = "https://bsky.social" + session = None + + def __init__( + self, + identifier: str = BLUESKY_IDENTIFIER, + app_password: str = BLUESKY_APP_PASSWORD, + service: str = "https://bsky.social", + ): + self.identifier = identifier + self.app_password = app_password + self.service = service + + def login(self): + """ + Login to Bluesky + """ + auth_response = requests.post( + f"{self.service}/xrpc/com.atproto.server.createSession", + json={"identifier": self.identifier, "password": self.app_password}, + ) + auth_response.raise_for_status() + self.session = auth_response.json() + + def get_profile(self, handle: str): + if not self.session: + self.login() + profile_response = None + try: + profile_response = requests.get( + f"{self.service}/xrpc/app.bsky.actor.getProfile", + headers={"Authorization": f'Bearer {self.session["accessJwt"]}'}, + params={"actor": handle}, + ) + profile_response.raise_for_status() + except Exception as exc: + print( + "Bluesky profile fetch failed with HTTP error: %s" % exc + ) # pragma: no cover + return None + + return profile_response.json() + + def get_avatar(self, handle: str): + profile = self.get_profile(handle) + return profile["avatar"] if profile else None def random_string(length=10): diff --git a/ivatar/views.py b/ivatar/views.py index bf4394d..2ad3466 100644 --- a/ivatar/views.py +++ b/ivatar/views.py @@ -5,7 +5,7 @@ views under / from io import BytesIO from os import path import hashlib -from urllib.request import urlopen +from ivatar.utils import urlopen from urllib.error import HTTPError, URLError from ssl import SSLError from django.views.generic.base import TemplateView, View @@ -36,8 +36,6 @@ from .ivataraccount.models import Photo from .ivataraccount.models import pil_format, file_format from .utils import is_trusted_url, mm_ng, resize_animated_gif -URL_TIMEOUT = 5 # in seconds - def get_size(request, size=DEFAULT_AVATAR_SIZE): """ @@ -396,7 +394,7 @@ class GravatarProxyView(View): # print("Cached Gravatar response: Default.") return redir_default(default) try: - urlopen(gravatar_test_url, timeout=URL_TIMEOUT) + urlopen(gravatar_test_url) except HTTPError as exc: if exc.code == 404: cache.set(gravatar_test_url, "default", 60) @@ -415,7 +413,7 @@ class GravatarProxyView(View): print("Cached Gravatar fetch failed with URL error: %s" % gravatar_url) return redir_default(default) - gravatarimagedata = urlopen(gravatar_url, timeout=URL_TIMEOUT) + gravatarimagedata = urlopen(gravatar_url) except HTTPError as exc: if exc.code != 404 and exc.code != 503: print(