fix: block TRACE/TRACK methods

common/helpers/malicious_requests.py

@@ -1,54 +1,73 @@
 import re
+import time
 from django.conf import settings
-from django.core.exceptions import MiddlewareNotUsed, SuspiciousOperation
+from django.http import HttpResponseBadRequest
+
+# TRACE and TRACK have no legitimate use in a web application and are
+# commonly used as cheap flood vectors (TRACE also enables Cross-Site Tracing).
+_ALWAYS_BLOCKED_METHODS = frozenset({'TRACE', 'TRACK'})
+
+# Caps repeated log lines for the same block reason during a flood, where
+# per-request logging itself becomes a source of latency/CPU load.
+_LOG_THROTTLE_SECONDS = 30
 
 
 class MaliciousRequestsMiddleware:
     def __init__(self, get_response):
         self.get_response = get_response
-        # One-time configuration and initialization.
-        used = False
+        self._log_throttle_state = {}
         if settings.MALICIOUS_URL_PATTERNS is not None:
             url_patterns = settings.MALICIOUS_URL_PATTERNS.split(',')
             self.malicious_url_patterns = list(map(lambda pattern: re.compile(pattern, re.IGNORECASE), url_patterns))
-            used = True
         if settings.MALICIOUS_FWD_PATTERNS is not None:
             fwd_patterns = settings.MALICIOUS_FWD_PATTERNS.split(',')
             self.malicious_fwd_patterns = list(map(lambda pattern: re.compile(pattern, re.IGNORECASE), fwd_patterns))
-            used = True
 
-        if not used:
-            raise MiddlewareNotUsed
+    def check_request_method(self, request):
+        if request.method in _ALWAYS_BLOCKED_METHODS:
+            self.log_filter_action(f'Blocking disallowed HTTP method "{request.method}"', key=f'method:{request.method}')
+            return HttpResponseBadRequest("Disallowed HTTP method")
 
     def check_request_url(self, request):
         path = request.get_full_path()
         for pattern in self.malicious_url_patterns:
             if pattern.search(path) is not None:
-                self.log_filter_action(f'Filtering malicious url "{path}" matching pattern "{pattern.pattern}"')
-                raise SuspiciousOperation("Malicious url detected")
+                self.log_filter_action(f'Filtering malicious url "{path}" matching pattern "{pattern.pattern}"', key=f'url:{pattern.pattern}')
+                return HttpResponseBadRequest("Malicious url detected")
 
     def check_request_fwd(self, request):
         fwd = request.headers['X-Forwarded-For'] if 'X-Forwarded-For' in request.headers else request.META['REMOTE_ADDR']
         for pattern in self.malicious_fwd_patterns:
             if pattern.search(fwd) is not None:
-                self.log_filter_action(f'Filtering malicious fwd "{fwd}" matching pattern "{pattern.pattern}"')
-                raise SuspiciousOperation("Malicious fwd detected")
+                self.log_filter_action(f'Filtering malicious fwd "{fwd}" matching pattern "{pattern.pattern}"', key=f'fwd:{pattern.pattern}')
+                return HttpResponseBadRequest("Malicious fwd detected")
 
-    @staticmethod
-    def log_filter_action(log_msg):
-        print(f'[MaliciousRequestsMiddleware] {log_msg}')
+    def log_filter_action(self, log_msg, key):
+        now = time.monotonic()
+        last_logged, suppressed = self._log_throttle_state.get(key, (None, 0))
+        if last_logged is not None and now - last_logged < _LOG_THROTTLE_SECONDS:
+            self._log_throttle_state[key] = (last_logged, suppressed + 1)
+            return
+        if suppressed:
+            print(f'[MaliciousRequestsMiddleware] {log_msg} (+{suppressed} more suppressed in last {_LOG_THROTTLE_SECONDS}s)')
+        else:
+            print(f'[MaliciousRequestsMiddleware] {log_msg}')
+        self._log_throttle_state[key] = (now, 0)
 
     def __call__(self, request):
-        # Code to be executed for each request before
-        # the view (and later middleware) are called.
-
-        hasattr(self, 'malicious_url_patterns') and self.check_request_url(request)
-        hasattr(self, 'malicious_fwd_patterns') and self.check_request_fwd(request)
+        response = self.check_request_method(request)
+        if response is not None:
+            return response
 
-        response = self.get_response(request)
+        if hasattr(self, 'malicious_url_patterns'):
+            response = self.check_request_url(request)
+            if response is not None:
+                return response
 
-        # Code to be executed for each request/response after
-        # the view is called.
+        if hasattr(self, 'malicious_fwd_patterns'):
+            response = self.check_request_fwd(request)
+            if response is not None:
+                return response
 
-        return response
+        return self.get_response(request)
 

common/tests/test_malicious_requests_middleware.py

@@ -0,0 +1,161 @@
+from unittest.mock import MagicMock, patch
+from django.test import TestCase, override_settings
+
+from common.helpers.malicious_requests import (
+    MaliciousRequestsMiddleware,
+    _ALWAYS_BLOCKED_METHODS,
+    _LOG_THROTTLE_SECONDS,
+)
+
+
+def _make_request(method='GET', path='/', fwd=None):
+    request = MagicMock()
+    request.method = method
+    request.get_full_path.return_value = path
+    request.META = {'REMOTE_ADDR': '127.0.0.1'}
+    if fwd is not None:
+        request.headers = {'X-Forwarded-For': fwd}
+    else:
+        request.headers = {}
+    return request
+
+
+def _middleware(get_response=None):
+    return MaliciousRequestsMiddleware(get_response or MagicMock(return_value=None))
+
+
+@override_settings(MALICIOUS_URL_PATTERNS=None, MALICIOUS_FWD_PATTERNS=None)
+class AlwaysBlockedMethodsTests(TestCase):
+    def test_always_blocked_set_contents(self):
+        self.assertIn('TRACE', _ALWAYS_BLOCKED_METHODS)
+        self.assertIn('TRACK', _ALWAYS_BLOCKED_METHODS)
+
+    def test_trace_is_blocked(self):
+        mw = _middleware()
+        response = mw(MagicMock(method='TRACE', get_full_path=MagicMock(return_value='/')))
+        self.assertEqual(response.status_code, 400)
+
+    def test_track_is_blocked(self):
+        mw = _middleware()
+        response = mw(MagicMock(method='TRACK', get_full_path=MagicMock(return_value='/')))
+        self.assertEqual(response.status_code, 400)
+
+    def test_get_is_allowed(self):
+        mw = _middleware()
+        req = _make_request('GET')
+        mw(req)  # should not raise
+
+    def test_post_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('POST'))
+
+    def test_put_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('PUT'))
+
+    def test_delete_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('DELETE'))
+
+    def test_patch_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('PATCH'))
+
+    def test_options_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('OPTIONS'))
+
+    def test_head_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request('HEAD'))
+
+
+@override_settings(MALICIOUS_URL_PATTERNS=r'\.php$,/wp-admin', MALICIOUS_FWD_PATTERNS=None)
+class UrlPatternTests(TestCase):
+    def test_malicious_url_is_blocked(self):
+        mw = _middleware()
+        response = mw(_make_request(path='/index.php'))
+        self.assertEqual(response.status_code, 400)
+
+    def test_another_malicious_url_is_blocked(self):
+        mw = _middleware()
+        response = mw(_make_request(path='/wp-admin/login'))
+        self.assertEqual(response.status_code, 400)
+
+    def test_clean_url_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request(path='/projects/123'))
+
+
+@override_settings(MALICIOUS_URL_PATTERNS=None, MALICIOUS_FWD_PATTERNS=r'^10\.')
+class FwdPatternTests(TestCase):
+    def test_malicious_fwd_is_blocked(self):
+        mw = _middleware()
+        response = mw(_make_request(fwd='10.0.0.1'))
+        self.assertEqual(response.status_code, 400)
+
+    def test_malicious_fwd_header_takes_precedence_over_remote_addr(self):
+        mw = _middleware()
+        response = mw(_make_request(fwd='10.1.2.3'))
+        self.assertEqual(response.status_code, 400)
+
+    def test_clean_fwd_is_allowed(self):
+        mw = _middleware()
+        mw(_make_request(fwd='203.0.113.5'))
+
+    def test_remote_addr_used_when_no_fwd_header(self):
+        mw = _middleware()
+        mw(_make_request())  # REMOTE_ADDR is 127.0.0.1, should not match ^10\.
+
+
+@override_settings(MALICIOUS_URL_PATTERNS=None, MALICIOUS_FWD_PATTERNS=None)
+class LogThrottlingTests(TestCase):
+    def _trace_request(self):
+        return MagicMock(method='TRACE', get_full_path=MagicMock(return_value='/'))
+
+    @patch('common.helpers.malicious_requests.print')
+    @patch('common.helpers.malicious_requests.time.monotonic')
+    def test_repeated_blocks_within_window_log_once(self, mock_monotonic, mock_print):
+        mock_monotonic.return_value = 0
+        mw = _middleware()
+
+        for _ in range(5):
+            response = mw(self._trace_request())
+            self.assertEqual(response.status_code, 400)
+
+        self.assertEqual(mock_print.call_count, 1)
+
+    @patch('common.helpers.malicious_requests.print')
+    @patch('common.helpers.malicious_requests.time.monotonic')
+    def test_block_after_window_logs_again_with_suppressed_count(self, mock_monotonic, mock_print):
+        mw = _middleware()
+
+        mock_monotonic.return_value = 0
+        response = mw(self._trace_request())
+        self.assertEqual(response.status_code, 400)
+
+        mock_monotonic.return_value = 5
+        for _ in range(3):
+            response = mw(self._trace_request())
+            self.assertEqual(response.status_code, 400)
+
+        mock_monotonic.return_value = _LOG_THROTTLE_SECONDS + 1
+        response = mw(self._trace_request())
+        self.assertEqual(response.status_code, 400)
+
+        self.assertEqual(mock_print.call_count, 2)
+        second_call_msg = mock_print.call_args_list[1].args[0]
+        self.assertIn('+3 more suppressed', second_call_msg)
+
+    @patch('common.helpers.malicious_requests.print')
+    @patch('common.helpers.malicious_requests.time.monotonic')
+    def test_different_block_reasons_throttle_independently(self, mock_monotonic, mock_print):
+        mock_monotonic.return_value = 0
+        mw = _middleware()
+
+        response = mw(self._trace_request())
+        self.assertEqual(response.status_code, 400)
+        response = mw(MagicMock(method='TRACK', get_full_path=MagicMock(return_value='/')))
+        self.assertEqual(response.status_code, 400)
+
+        self.assertEqual(mock_print.call_count, 2)

democracylab/logging.py

@@ -18,7 +18,7 @@ def dump_request_summary(request):
     user = (hasattr(request, 'user') and request.user.is_authenticated and request.user.username) or ''
     url = request.path
     method = request.method
-    body = censor_sensitive_fields(dict(getattr(request, method)))
+    body = censor_sensitive_fields(dict(getattr(request, method, None) or {}))
 
     return '({user}) {method} {url} {body}'.format(user=user, url=url, method=method, body=body)