"""
Middleware to block clients that trigger excessive 404 responses.
"""
import logging
from collections.abc import Awaitable
from http import HTTPStatus
from django.core.handlers.wsgi import WSGIRequest
from django.http import HttpResponseForbidden
from django.http.request import HttpRequest
from django.http.response import HttpResponseBase
from smarter.common.const import SMARTER_CUSTOMER_SUPPORT_EMAIL
from smarter.common.helpers.console_helpers import formatted_text
from smarter.common.mixins import SmarterMiddlewareMixin
from smarter.common.utils import is_authenticated_request
from smarter.lib.cache import lazy_cache as cache
from smarter.lib.django import waffle
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return (waffle.switch_is_active(SmarterWaffleSwitches.MIDDLEWARE_LOGGING)) or level >= logging.WARNING
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
logger.debug("Loading %s", formatted_text(__name__ + ".SmarterBlockExcessive404Middleware"))
[docs]
class SmarterBlockExcessive404Middleware(SmarterMiddlewareMixin):
"""
Middleware to block unauthenticated clients that trigger excessive 404 responses.
This is a countermeasure against abusive or automated 'bot' clients probing for non-existent resources.
This middleware monitors incoming HTTP requests and tracks the number of 404 (Not Found)
responses generated by each client IP address. If a client exceeds a configurable threshold
of 404 responses within a specified time window, further requests from that client will be
blocked with a 403 Forbidden response for the remainder of the timeout period.
The middleware is designed to help mitigate abusive or automated clients that probe for
non-existent resources, which can be indicative of malicious activity or misconfigured bots.
:cvar int THROTTLE_LIMIT: The maximum number of allowed 404 responses from a single client IP within the timeout period before blocking is triggered. Default is 25.
:cvar int THROTTLE_TIMEOUT: The duration of the timeout window in seconds during which 404 responses are counted and blocking is enforced. Default is 600 seconds (10 minutes).
.. note::
- Authenticated users are exempt from this blocking mechanism.
- The client IP is determined using the :meth:`get_client_ip` method.
- The 404 count and blocking state are stored in the Django cache backend.
- Logging is performed using a waffle switch to control verbosity.
**Example**
To enable this middleware, add it to your Django project's middleware settings::
MIDDLEWARE = [
...
'smarter.lib.django.middleware.excessive_404.SmarterBlockExcessive404Middleware',
...
]
"""
THROTTLE_LIMIT = 25
"""The maximum number of allowed 404 responses from a single unauthenticated client IP within the timeout period before blocking is triggered."""
THROTTLE_TIMEOUT = 600 # seconds (10 minutes)
"""The duration of the timeout window in seconds during which 404 responses are counted and blocking is enforced."""
@property
def formatted_class_name(self) -> str:
"""Return the formatted class name for logging purposes."""
return formatted_text(f"{__name__}.{SmarterBlockExcessive404Middleware.__name__}")
[docs]
def process_response(self, request: WSGIRequest, response):
"""
Process the HTTP response and apply excessive 404 blocking logic.
:param request: The incoming HTTP request object.
:type request: django.core.handlers.wsgi.WSGIRequest
:param response: The outgoing HTTP response object.
:type response: django.http.HttpResponse
:returns: The original response, or a 403 Forbidden response if the client has exceeded the allowed number of 404 responses.
:rtype: django.http.HttpResponse
"""
if not waffle.switch_is_active(SmarterWaffleSwitches.ENABLE_MIDDLEWARE_EXCESSIVE_404):
return response
# skip if the response is anything other than a 404
if response.status_code != HTTPStatus.NOT_FOUND:
return response
# skip this for authenticated users
if is_authenticated_request(request):
return response
logger.debug("%s.process_response(): %s", self.formatted_class_name, self.smarter_build_absolute_uri(request))
client_ip = self.get_client_ip(request)
if not client_ip:
return response
throttle_key = f"excessive_404_throttle:{client_ip}"
blocked_count = cache.get(throttle_key, 0)
if blocked_count >= self.THROTTLE_LIMIT:
logger.warning("%s Throttled client %s after %d 404s", self.formatted_class_name, client_ip, blocked_count)
return HttpResponseForbidden(
f"You have been blocked due to too many invalid requests from your IP. Try again later or contact {SMARTER_CUSTOMER_SUPPORT_EMAIL}."
)
try:
blocked_count = cache.incr(throttle_key)
except ValueError:
cache.set(throttle_key, 1, timeout=self.THROTTLE_TIMEOUT)
blocked_count = 1
else:
cache.set(throttle_key, blocked_count, timeout=self.THROTTLE_TIMEOUT)
return response