add ability to block ip addresses that are submitting spam OWASP-BLT#…

…1938

blt/middleware/count_ip_requests.py

@@ -0,0 +1,24 @@
+from django.utils.deprecation import MiddlewareMixin
+from user_agents import parse
+
+from website.models import IP
+
+
+class MonitorIPMiddleware(MiddlewareMixin):
+    def process_request(self, request):
+        x_forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR")
+        if x_forwarded_for:
+            ip = x_forwarded_for.split(",")[0].strip()
+        else:
+            ip = request.META.get("REMOTE_ADDR")
+
+        user_agent = request.META.get("HTTP_USER_AGENT", "")
+        parsed_agent = parse(user_agent)
+
+        if ip:
+            ip_record = IP.objects.filter(address=ip).first()
+
+            if ip_record:
+                ip_record.user_agent_string = parsed_agent
+                ip_record.count += 1
+                ip_record.save()

blt/middleware/ip_restrict.py

@@ -0,0 +1,70 @@
+import ipaddress
+
+from django.core.cache import cache
+from django.http import HttpResponseForbidden
+from user_agents import parse
+
+from website.models import BlockedIP
+
+
+class IPRestrictMiddleware:
+    """
+    Middleware to restrict access based on client IP addresses.
+    """
+
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def blocked_ips(self):
+        blocked_ips = cache.get("blocked_ips")
+        if blocked_ips is None:
+            blocked_addresses = BlockedIP.objects.values_list("address", flat=True)
+            blocked_ips = set(blocked_addresses)
+            cache.set("blocked_ips", blocked_ips, timeout=86400)
+        return blocked_ips
+
+    def blocked_ip_ranges(self):
+        blocked_ip_ranges = cache.get("blocked_ip_ranges")
+        if blocked_ip_ranges is None:
+            blocked_ip_start = BlockedIP.objects.values_list("address_range_start", flat=True)
+            blocked_ip_end = BlockedIP.objects.values_list("address_range_end", flat=True)
+            blocked_ip_ranges = list(zip(blocked_ip_start, blocked_ip_end))
+            cache.set("blocked_ip_ranges", blocked_ip_ranges, timeout=86400)
+        return blocked_ip_ranges
+
+    def ip_in_range(self, ip, ip_ranges):
+        ip_int = int(ipaddress.IPv4Address(ip))
+        for start, end in ip_ranges:
+            start_int = int(ipaddress.IPv4Address(start))
+            end_int = int(ipaddress.IPv4Address(end))
+            if start_int <= ip_int <= end_int:
+                return True
+        return False
+
+    def blocked_agents(self):
+        blocked_agents = cache.get("blocked_agents")
+        if blocked_agents is None:
+            blocked_user_agents = BlockedIP.objects.values_list("user_agent_string", flat=True)
+            blocked_agents = set(blocked_user_agents)
+            cache.set("blocked_agents", blocked_agents, timeout=86400)
+        return blocked_agents
+
+    def __call__(self, request):
+        ip = request.META.get("REMOTE_ADDR")
+        user_agent = request.META.get("HTTP_USER_AGENT", "")
+        parsed_agent = parse(user_agent)
+
+        if ip:
+            if ip in self.blocked_ips():
+                return HttpResponseForbidden(
+                    "Your IP address is restricted from accessing this site."
+                )
+            blocked_ip_ranges = self.blocked_ip_ranges()
+            if self.ip_in_range(ip, blocked_ip_ranges):
+                return HttpResponseForbidden(
+                    "Your IP address is restricted from accessing this site."
+                )
+        if parsed_agent and parsed_agent in self.blocked_agents():
+            return HttpResponseForbidden("Your IP address is restricted from accessing this site.")
+
+        return self.get_response(request)

blt/settings.py

@@ -118,6 +118,8 @@
     "django.middleware.security.SecurityMiddleware",
     "whitenoise.middleware.WhiteNoiseMiddleware",
     "tz_detect.middleware.TimezoneMiddleware",
+    "blt.middleware.ip_restrict.IPRestrictMiddleware",
+    "blt.middleware.count_ip_requests.MonitorIPMiddleware",
 )
 
 TESTING = len(sys.argv) > 1 and sys.argv[1] == "test"
@@ -360,7 +362,7 @@
         },
     },
 }
-
+DEFAULT_FILE_STORAGE = "django.core.files.storage.FileSystemStorage"
 USERS_AVATAR_PATH = "avatars"
 AVATAR_PATH = os.path.join(MEDIA_ROOT, USERS_AVATAR_PATH)
 

website/admin.py

@@ -7,6 +7,7 @@
 from website.models import (
     IP,
     Bid,
+    BlockedIP,
     ChatBotLog,
     Company,
     CompanyAdmin,
@@ -276,9 +277,49 @@ def issue_description(self, obj):
         return obj.issue.description
 
 
+def block_ip(modeladmin, request, queryset):
+    for ip in queryset:
+        BlockedIP.objects.create(address=ip.address, count=ip.count)
+    modeladmin.message_user(request, "Selected IPs have been blocked successfully.")
+
+
+block_ip.short_description = "Block selected IPs"
+
+
+def unblock_ip(modeladmin, request, queryset):
+    for ip in queryset:
+        BlockedIP.objects.filter(ip=ip.address).delete()
+    modeladmin.message_user(request, "Selected IPs have ben unblocked successfully")
+
+
+unblock_ip.short_description = "Unblock selected IPs"
+
+
+def block_user_agent(modeladmin, request, queryset):
+    for ip in queryset:
+        BlockedIP.objects.create(user_agent_string=ip.user_agent_string)
+
+    modeladmin.message_user(request, "Selected UserAgent have been blocked successfully.")
+
+
+block_user_agent.short_description = "Block selected UserAgent"
+
+
+def unblock_user_agent(modeladmin, request, queryset):
+    for ip in queryset:
+        BlockedIP.objects.filter(user_agent_string=ip.user_agent_string).delete()
+
+    modeladmin.message_user(request, "Selected UserAgent have been unblocked successfully.")
+
+
+unblock_user_agent.short_description = "Unblock selected UserAgent"
+
+
 class IPAdmin(admin.ModelAdmin):
     list_display = ("id", "address", "user", "issuenumber", "created", "agent", "path")
 
+    actions = [block_ip, unblock_ip, block_user_agent, unblock_user_agent]
+
 
 class MonitorAdmin(admin.ModelAdmin):
     list_display = (
@@ -305,6 +346,17 @@ class SuggestionVotesAdmin(admin.ModelAdmin):
     list_display = ("user", "suggestion", "up_vote", "down_vote")
 
 
+class BlockedIPAdmin(admin.ModelAdmin):
+    list_display = (
+        "address",
+        "reason_for_block",
+        "address_range_start",
+        "address_range_end",
+        "user_agent_string",
+        "count",
+    )
+
+
 class ProjectAdmin(admin.ModelAdmin):
     list_display = (
         "id",
@@ -315,7 +367,6 @@ class ProjectAdmin(admin.ModelAdmin):
         "created",
         "modified",
     )
-
     search_fields = ["name", "description", "slug"]
 
 
@@ -338,6 +389,7 @@ class ProjectAdmin(admin.ModelAdmin):
 admin.site.register(IssueScreenshot, IssueScreenshotAdmin)
 admin.site.register(HuntPrize)
 admin.site.register(ChatBotLog, ChatBotLogAdmin)
+admin.site.register(BlockedIP, BlockedIPAdmin)
 admin.site.register(Suggestion, SuggestionAdmin)
 admin.site.register(SuggestionVotes, SuggestionVotesAdmin)
 

website/migrations/0125_blockedip.py

@@ -0,0 +1,44 @@
+# Generated by Django 5.1 on 2024-08-09 19:59
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("website", "0124_company_tags_domain_tags_project_tags_and_more"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="BlockedIP",
+            fields=[
+                (
+                    "id",
+                    models.AutoField(
+                        auto_created=True,
+                        primary_key=True,
+                        serialize=False,
+                        verbose_name="ID",
+                    ),
+                ),
+                ("address", models.GenericIPAddressField(blank=True, null=True)),
+                (
+                    "reason_for_block",
+                    models.TextField(blank=True, max_length=255, null=True),
+                ),
+                (
+                    "address_range_start",
+                    models.GenericIPAddressField(blank=True, null=True),
+                ),
+                (
+                    "address_range_end",
+                    models.GenericIPAddressField(blank=True, null=True),
+                ),
+                (
+                    "user_agent_string",
+                    models.CharField(blank=True, default="", max_length=255, null=True),
+                ),
+                ("count", models.IntegerField(default=1)),
+            ],
+        ),
+    ]

website/models.py

@@ -786,3 +786,15 @@ class BaconToken(models.Model):
 
     def __str__(self):
         return f"{self.user.username} - {self.amount} BACON"
+
+
+class BlockedIP(models.Model):
+    address = models.GenericIPAddressField(null=True, blank=True)
+    reason_for_block = models.TextField(blank=True, null=True, max_length=255)
+    address_range_start = models.GenericIPAddressField(null=True, blank=True)
+    address_range_end = models.GenericIPAddressField(null=True, blank=True)
+    user_agent_string = models.CharField(max_length=255, default="", null=True, blank=True)
+    count = models.IntegerField(default=1)
+
+    def __str__(self):
+        return f"user agent : {self.user_agent_string} | IP : {self.address}"