Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions firewall/middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ def __init__(self, get_response):
self.get_response = get_response

def __call__(self, request):
ip = None

if self.firewall_on:

Expand Down Expand Up @@ -74,11 +75,11 @@ def __call__(self, request):
request.firewall = self # pass this on the request so it can be used later
response = self.get_response(request)

if self.firewall_on:
if ip and self.firewall_on:
# did we get a 404? If so, see if this ip is messing with up
if response.status_code == 404:
if ip in self.probation_ips:
self.probation_ips[ip].append([datetime.now()])
self.probation_ips[ip].append(datetime.now())
if len(self.probation_ips[ip]) > 2:
time_since_first = datetime.now() - self.probation_ips[ip][0]
if time_since_first.seconds < PROBATION_LIMIT:
Expand All @@ -92,7 +93,7 @@ def __call__(self, request):
return HttpResponseForbidden()
else:
# just pop the first one
self.probation_ips[ip].popitem(last=False)
self.probation_ips[ip] = self.probation_ips[ip][1:]
else:
self.probation_ips[ip] = [datetime.now()]
elif ip in self.probation_ips:
Expand All @@ -118,4 +119,4 @@ def get_user_ip(self, request):
# ip = request.META.get('REMOTE_ADDR')
# return ip
ip, _ = self.ipw.get_client_ip(request.META)
return ip
return ip
36 changes: 28 additions & 8 deletions firewall/tests.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,29 +2,49 @@
from django.urls import reverse
from freezegun import freeze_time
from datetime import datetime, timedelta
from member.models import Member

# Create your tests here.

class FirewallTests(TestCase):

def test_ip_firewall(self):

c = Client()
response = c.get(f"/xxx")
c.force_login(Member.objects.create_user(email='super@b.c', is_superuser=True))
_ = c.get("/firewall/firewall_on")
_ = c.get("/firewall/reset_stats")

response = c.get("/111")
assert response.status_code==404
response = c.get(f"/xxx")
response = c.get("/222")
assert response.status_code==404
response = c.get(f"/xxx") # third time should get permission denied
response = c.get("/333") # third time should get permission denied
assert response.status_code==403

response = c.get(f"/xxx") # fourth time should get permission denied
response = c.get("/444") # fourth time should get permission denied
assert response.status_code==403

with freeze_time(datetime.now()+timedelta(seconds=100)):
response = c.get(f"/xxx") # 100 seconds in future should get permission denied
response = c.get("/555") # 100 seconds in future should get permission denied
assert response.status_code==403

with freeze_time(datetime.now()+timedelta(seconds=601)):
response = c.get(f"/xxx") # 600 seconds in future should get not found again
with freeze_time(datetime.now()+timedelta(seconds=600)):
response = c.get("/666") # 600 seconds in future should get not found again
assert response.status_code==404

def test_clear_probation(self):
c = Client()
c.force_login(Member.objects.create_user(email='super@b.c', is_superuser=True))
_ = c.get("/firewall/firewall_on")
_ = c.get("/firewall/reset_stats")

response = c.get('/xxx')
assert response.status_code==404
response = c.get(f"/xxx")
assert response.status_code==404

# if we don't send another until the future, we should come back 404 b/c the first
# request won't count against us anymore.
with freeze_time(datetime.now()+timedelta(seconds=601)):
response = c.get(f"/xxx")
assert response.status_code==404
2 changes: 1 addition & 1 deletion go3/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -355,4 +355,4 @@
}
}

START_FIREWALL = env('START_FIREWALL',default=True)
START_FIREWALL = env('START_FIREWALL',default=not _testing)
Loading