Coverage for ivatar/test_security_fixes.py: 100%

1""" 

2Test security fixes for ETag sanitization and URL validation 

3 

4This test suite covers two critical security fixes: 

5 

61. ETag Header Sanitization (middleware.py): 

7 - Prevents BadHeaderError when hash values contain newlines or control characters 

8 - Sanitizes ETag values to remove potentially malicious characters 

9 - Maintains functionality for normal hash values 

10 

112. URL Validation (utils.py): 

12 - Gracefully handles malformed URLs with control characters 

13 - Converts http.client.InvalidURL exceptions to URLError for consistent handling 

14 - Logs potential injection attempts for security monitoring 

15 - Maintains compatibility with existing error handling code 

16 

17These fixes address real-world attack scenarios including: 

18- Header injection via newlines in avatar hashes 

19- SQL injection attempts in URL parameters 

20- Control character injection in URLs 

21""" 

22 

23import os 

24import django 

25from django.test import TestCase, RequestFactory 

26from django.http import HttpResponse 

27from unittest.mock import patch, Mock 

28import http.client 

29from urllib.error import URLError 

30 

31os.environ["DJANGO_SETTINGS_MODULE"] = "ivatar.settings" 

32django.setup() 

33 

34from ivatar.middleware import CustomLocaleMiddleware 

35from ivatar.utils import urlopen 

36 

37 

38class ETagSanitizationTest(TestCase): 

39 """ 

40 Test ETag header sanitization in middleware 

41 """ 

42 

43 def setUp(self): 

44 self.factory = RequestFactory() 

45 self.middleware = CustomLocaleMiddleware(lambda request: HttpResponse()) 

46 

47 def test_etag_with_newlines_sanitized(self): 

48 """Test that ETag values with newlines are properly sanitized""" 

49 # Create a request for an avatar URL with a hash containing newlines 

50 request = self.factory.get("/avatar/404\n") 

51 response = HttpResponse() 

52 

53 # Process the response through middleware 

54 processed_response = self.middleware.process_response(request, response) 

55 

56 # Check that ETag is set and doesn't contain newlines 

57 self.assertIn("Etag", processed_response) 

58 etag_value = processed_response["Etag"] 

59 self.assertNotIn("\n", etag_value) 

60 self.assertNotIn("\r", etag_value) 

61 self.assertEqual(etag_value, '"404"') 

62 

63 def test_etag_with_carriage_return_sanitized(self): 

64 """Test that ETag values with carriage returns are properly sanitized""" 

65 request = self.factory.get("/avatar/test\r\ninjection") 

66 response = HttpResponse() 

67 

68 processed_response = self.middleware.process_response(request, response) 

69 

70 etag_value = processed_response["Etag"] 

71 self.assertNotIn("\n", etag_value) 

72 self.assertNotIn("\r", etag_value) 

73 self.assertEqual(etag_value, '"testinjection"') 

74 

75 def test_etag_with_control_characters_sanitized(self): 

76 """Test that ETag values with control characters are properly sanitized""" 

77 request = self.factory.get("/avatar/test\x00\x01control") 

78 response = HttpResponse() 

79 

80 processed_response = self.middleware.process_response(request, response) 

81 

82 etag_value = processed_response["Etag"] 

83 # Control characters should be removed 

84 self.assertEqual(etag_value, '"testcontrol"') 

85 

86 def test_etag_normal_hash_unchanged(self): 

87 """Test that normal hash values are unchanged""" 

88 request = self.factory.get("/avatar/c1923131dec28fa7d41356cfb15edd2b") 

89 response = HttpResponse() 

90 

91 processed_response = self.middleware.process_response(request, response) 

92 

93 etag_value = processed_response["Etag"] 

94 self.assertEqual(etag_value, '"c1923131dec28fa7d41356cfb15edd2b"') 

95 

96 def test_etag_fallback_for_short_path(self): 

97 """Test ETag fallback when path is too short""" 

98 request = self.factory.get("/avatar/") 

99 response = HttpResponse(b"test content") 

100 

101 processed_response = self.middleware.process_response(request, response) 

102 

103 # Should use content hash as fallback 

104 self.assertIn("Etag", processed_response) 

105 etag_value = processed_response["Etag"] 

106 self.assertTrue(etag_value.startswith('"') and etag_value.endswith('"')) 

107 

108 def test_non_avatar_urls_unchanged(self): 

109 """Test that non-avatar URLs are processed normally by parent middleware""" 

110 request = self.factory.get("/some/other/path") 

111 response = HttpResponse() 

112 

113 # Mock the parent's process_response 

114 with patch.object( 

115 CustomLocaleMiddleware.__bases__[0], "process_response" 

116 ) as mock_parent: 

117 mock_parent.return_value = response 

118 self.middleware.process_response(request, response) 

119 mock_parent.assert_called_once_with(request, response) 

120 

121 

122class URLValidationTest(TestCase): 

123 """ 

124 Test URL validation and error handling in urlopen function 

125 """ 

126 

127 @patch("ivatar.utils.urlopen_orig") 

128 def test_invalid_url_handling(self, mock_urlopen_orig): 

129 """Test that InvalidURL exceptions are handled gracefully""" 

130 # Simulate http.client.InvalidURL exception 

131 mock_urlopen_orig.side_effect = http.client.InvalidURL( 

132 "URL can't contain control characters" 

133 ) 

134 

135 with self.assertRaises(URLError) as context: 

136 urlopen("http://example.com/bad\x00url") 

137 

138 # Check that it was converted to URLError with appropriate message 

139 self.assertIn("Invalid URL", str(context.exception)) 

140 

141 @patch("ivatar.utils.urlopen_orig") 

142 def test_malformed_url_handling(self, mock_urlopen_orig): 

143 """Test that ValueError exceptions are handled gracefully""" 

144 mock_urlopen_orig.side_effect = ValueError("Invalid URL format") 

145 

146 with self.assertRaises(URLError) as context: 

147 urlopen("not-a-valid-url") 

148 

149 self.assertIn("Malformed URL", str(context.exception)) 

150 

151 @patch("ivatar.utils.urlopen_orig") 

152 def test_unicode_error_handling(self, mock_urlopen_orig): 

153 """Test that UnicodeError exceptions are handled gracefully""" 

154 mock_urlopen_orig.side_effect = UnicodeError("Unicode decode error") 

155 

156 with self.assertRaises(URLError) as context: 

157 urlopen("http://example.com/unicode-issue") 

158 

159 self.assertIn("Malformed URL", str(context.exception)) 

160 

161 @patch("ivatar.utils.urlopen_orig") 

162 def test_other_exceptions_passthrough(self, mock_urlopen_orig): 

163 """Test that other exceptions are passed through unchanged""" 

164 mock_urlopen_orig.side_effect = ConnectionError("Network error") 

165 

166 with self.assertRaises(ConnectionError): 

167 urlopen("http://example.com/") 

168 

169 @patch("ivatar.utils.urlopen_orig") 

170 def test_successful_url_request(self, mock_urlopen_orig): 

171 """Test that successful requests work normally""" 

172 mock_response = Mock() 

173 mock_urlopen_orig.return_value = mock_response 

174 

175 result = urlopen("http://example.com/") 

176 

177 self.assertEqual(result, mock_response) 

178 mock_urlopen_orig.assert_called_once() 

179 

180 @patch("ivatar.utils.logger") 

181 @patch("ivatar.utils.urlopen_orig") 

182 def test_security_logging(self, mock_urlopen_orig, mock_logger): 

183 """Test that security issues are properly logged""" 

184 mock_urlopen_orig.side_effect = http.client.InvalidURL( 

185 "URL can't contain control characters" 

186 ) 

187 

188 with self.assertRaises(URLError): 

189 urlopen("http://example.com/malicious\x00url") 

190 

191 # Check that security warning was logged 

192 mock_logger.warning.assert_called_once() 

193 log_call = mock_logger.warning.call_args[0][0] 

194 self.assertIn("Invalid URL detected", log_call) 

195 self.assertIn("possible injection attempt", log_call) 

196 

197 

198class IntegrationTest(TestCase): 

199 """ 

200 Integration tests for the security fixes 

201 """ 

202 

203 def test_sql_injection_attempt_url(self): 

204 """Test handling of the actual SQL injection URL from the error log""" 

205 malicious_path = ( 

206 "/avatar/c1923131dec28fa7d41356cfb15edd2b?s=80&d=mm'; DROP TABLE .; --" 

207 ) 

208 

209 # Test middleware ETag handling 

210 factory = RequestFactory() 

211 request = factory.get(malicious_path) 

212 middleware = CustomLocaleMiddleware(lambda request: HttpResponse()) 

213 response = HttpResponse() 

214 

215 processed_response = middleware.process_response(request, response) 

216 

217 # Should handle the hash part safely 

218 self.assertIn("Etag", processed_response) 

219 etag_value = processed_response["Etag"] 

220 # The hash part should be extracted correctly (before the ?) 

221 self.assertEqual(etag_value, '"c1923131dec28fa7d41356cfb15edd2b"') 

222 

223 def test_newline_injection_attempt(self): 

224 """Test handling of newline injection in avatar hash""" 

225 malicious_path = "/avatar/404\nInjected-Header: malicious" 

226 

227 factory = RequestFactory() 

228 request = factory.get(malicious_path) 

229 middleware = CustomLocaleMiddleware(lambda request: HttpResponse()) 

230 response = HttpResponse() 

231 

232 processed_response = middleware.process_response(request, response) 

233 

234 # Should sanitize the newline 

235 etag_value = processed_response["Etag"] 

236 self.assertEqual(etag_value, '"404Injected-Header: malicious"') 

237 self.assertNotIn("\n", etag_value)