Coverage for ivatar/middleware.py: 86%

1""" 

2Middleware classes 

3""" 

4 

5import hashlib 

6from django.utils.deprecation import MiddlewareMixin 

7from django.middleware.locale import LocaleMiddleware 

8from django.utils.cache import get_conditional_response 

9from django.utils.http import parse_http_date_safe 

10 

11 

12class CustomLocaleMiddleware(LocaleMiddleware): 

13 """ 

14 Middleware that extends LocaleMiddleware to skip Vary header processing for image URLs 

15 and handle conditional requests (ETag / If-None-Match -> 304 Not Modified). 

16 """ 

17 

18 def process_response(self, request, response): 

19 # Check if this is an image-related URL 

20 path = request.path 

21 if any( 

22 path.startswith(prefix) 

23 for prefix in [ 

24 "/avatar/", 

25 "/gravatarproxy/", 

26 "/blueskyproxy/", 

27 "/accounts/raw_image/", 

28 ] 

29 ): 

30 # Preserve existing ETag from view if present, otherwise compute fallback 

31 etag = response.get("Etag") 

32 if not etag: 

33 path_parts = path.strip("/").split("/") 

34 if len(path_parts) >= 2: 

35 hash_value = path_parts[1] # Get the hash part 

36 # Sanitize hash_value to remove newlines and other control characters 

37 # that would cause BadHeaderError 

38 hash_value = "".join( 

39 c for c in hash_value if c.isprintable() and c not in "\r\n" 

40 ) 

41 etag = f'"{hash_value}"' 

42 response["Etag"] = etag 

43 else: 

44 # Fallback to content hash if we can't extract from URL 

45 # Use hashlib.md5 for stable hash across processes 

46 content_hash = hashlib.md5(response.content).hexdigest() 

47 etag = f'"{content_hash}"' 

48 response["Etag"] = etag 

49 

50 # If response is 200 OK, evaluate conditional request headers (If-None-Match, If-Modified-Since) 

51 if response.status_code == 200: 

52 last_mod_header = response.get("Last-Modified") 

53 last_modified = ( 

54 parse_http_date_safe(last_mod_header) 

55 if last_mod_header 

56 else None 

57 ) 

58 conditional_response = get_conditional_response( 

59 request, 

60 etag=etag, 

61 last_modified=last_modified, 

62 response=response, 

63 ) 

64 if conditional_response is not response: 

65 conditional_response["Etag"] = etag 

66 if "Vary" in response: 

67 conditional_response["Vary"] = response["Vary"] 

68 return conditional_response 

69 

70 # Skip the parent's process_response to avoid adding Accept-Language to Vary 

71 return response 

72 

73 # For all other URLs, use the parent's behavior 

74 return super().process_response(request, response) 

75 

76 

77class MultipleProxyMiddleware( 

78 MiddlewareMixin 

79): # pylint: disable=too-few-public-methods 

80 """ 

81 Middleware to rewrite proxy headers for deployments 

82 with multiple proxies 

83 """ 

84 

85 def process_request(self, request): # pylint: disable=no-self-use 

86 """ 

87 Rewrites the proxy headers so that forwarded server is 

88 used if available. 

89 """ 

90 if "HTTP_X_FORWARDED_SERVER" in request.META: 

91 request.META["HTTP_X_FORWARDED_HOST"] = request.META[ 

92 "HTTP_X_FORWARDED_SERVER" 

93 ]