base setup
This commit is contained in:
0
at_django_boilerplate/core/middleware/__init__.py
Executable file
0
at_django_boilerplate/core/middleware/__init__.py
Executable file
67
at_django_boilerplate/core/middleware/block_ips.py
Executable file
67
at_django_boilerplate/core/middleware/block_ips.py
Executable file
@@ -0,0 +1,67 @@
|
||||
import time
|
||||
from django.conf import settings
|
||||
from django.http import HttpResponseForbidden
|
||||
from threading import Lock
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class BlockIPMiddleware:
|
||||
BLOCKED_IPS = {} # To store the IP and timestamp of the attempts
|
||||
MAX_ATTEMPTS = 25 # Maximum number of attempts before blocking
|
||||
BLOCK_DURATION = 30 # Block duration in seconds (e.g., 60 seconds = 1 minute)
|
||||
lock = Lock() # Lock to ensure thread safety
|
||||
WHITE_LIST_IPS = ['127.0.0.1']
|
||||
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
|
||||
def __call__(self, request):
|
||||
ip = self.get_client_ip(request)
|
||||
if ip in self.WHITE_LIST_IPS:
|
||||
return self.get_response(request)
|
||||
|
||||
with self.lock:
|
||||
# Check if IP is already blocked
|
||||
if ip in self.BLOCKED_IPS:
|
||||
attempts, block_time, blocked = self.BLOCKED_IPS[ip]
|
||||
if blocked:
|
||||
if (time.time() - block_time) < self.BLOCK_DURATION:
|
||||
logger.warning(f"Blocked IP attempt: {ip}")
|
||||
return HttpResponseForbidden("Your IP has been temporarily blocked due to multiple unsuccessful attempts.")
|
||||
else:
|
||||
del self.BLOCKED_IPS[ip]
|
||||
|
||||
# Process the request and get the response
|
||||
response = self.get_response(request)
|
||||
|
||||
with self.lock:
|
||||
# If response status code indicates an error (e.g., 4xx or 5xx), count it as an attempt
|
||||
if response.status_code >= 300:
|
||||
self.count_attempt(ip)
|
||||
else:
|
||||
# Reset the counter on a successful request
|
||||
if ip in self.BLOCKED_IPS:
|
||||
del self.BLOCKED_IPS[ip]
|
||||
|
||||
return response
|
||||
|
||||
def count_attempt(self, ip):
|
||||
attempts = self.BLOCKED_IPS.get(ip, [0, time.time(), False])
|
||||
attempts[0] += 1
|
||||
|
||||
# If attempts exceed MAX_ATTEMPTS, block the IP
|
||||
if attempts[0] >= self.MAX_ATTEMPTS:
|
||||
attempts[2] = True
|
||||
attempts[1] = time.time() # Update block time
|
||||
logger.warning(f"IP blocked: {ip}")
|
||||
|
||||
self.BLOCKED_IPS[ip] = attempts
|
||||
|
||||
def get_client_ip(self, request):
|
||||
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
||||
if x_forwarded_for:
|
||||
ip = x_forwarded_for.split(',')[0]
|
||||
else:
|
||||
ip = request.META.get('REMOTE_ADDR')
|
||||
return ip
|
||||
12
at_django_boilerplate/core/middleware/country_detection.py
Executable file
12
at_django_boilerplate/core/middleware/country_detection.py
Executable file
@@ -0,0 +1,12 @@
|
||||
from at_django_boilerplate.utils.geolocation import get_ip_and_country
|
||||
|
||||
|
||||
class CountryDetectionMiddleware:
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
|
||||
def __call__(self, request):
|
||||
ip, country,currency = get_ip_and_country(request)
|
||||
request.session['country'] = country
|
||||
request.session['currency'] = currency
|
||||
return self.get_response(request)
|
||||
Reference in New Issue
Block a user