Fixup IP logging

Makefile

@@ -51,3 +51,15 @@ clean-test:
 clean-dev-server:
 	docker-compose rm -sfv
 	docker volume rm -f core_postgres
+
+dev-logs:
+	docker-compose logs -f backend
+
+dev-sql:
+	docker-compose exec database psql -U postgres postgres
+
+dev-fake-bulk-data:
+	docker-compose exec backend python -m scripts.fake generate --teams 10000 --users 2 --categories 10 --challenges 100 --solves 10000
+
+dev-shell:
+	docker-compose exec backend ./src/manage.py shell

src/member/migrations/0010_fix_client_ip_addresses.py

@@ -0,0 +1,36 @@
+# Generated by Django 4.0.2 on 2022-05-01 19:51
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+def fix_ip_addresses(apps, schema_editor):
+    UserIP = apps.get_model('member', 'userip')
+    db_alias = schema_editor.connection.alias
+    for userip in UserIP.objects.using(db_alias).all():
+        userip.ip = userip.ip.split(",")[0]
+        userip.save()
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('member', '0009_alter_member_options_alter_member_username_and_more'),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            fix_ip_addresses,
+            # Marked as elidable as all new IP addresses will be in the correct format
+            elidable=True,
+        ),
+        migrations.AlterField(
+            model_name='userip',
+            name='ip',
+            field=models.GenericIPAddressField(),
+        ),
+        migrations.AlterField(
+            model_name='userip',
+            name='user',
+            field=models.ForeignKey(null=True, on_delete=models.deletion.SET_NULL,
+                                    related_name='ips', to=settings.AUTH_USER_MODEL),
+        ),
+    ]

src/member/models.py

@@ -80,8 +80,8 @@ def should_deny_admin(self):
 
 
 class UserIP(ExportModelOperationsMixin("user_ip"), models.Model):
-    user = models.ForeignKey(get_user_model(), on_delete=SET_NULL, null=True)
-    ip = models.CharField(max_length=255)
+    user = models.ForeignKey(get_user_model(), on_delete=SET_NULL, null=True, related_name="ips")
+    ip = models.GenericIPAddressField()
     seen = models.IntegerField(default=1)
     last_seen = models.DateTimeField(default=timezone.now)
     user_agent = models.CharField(max_length=255)
@@ -90,7 +90,7 @@ class UserIP(ExportModelOperationsMixin("user_ip"), models.Model):
     def hook(request):
         if not request.user.is_authenticated:
             return
-        ip = request.headers.get("x-forwarded-for", "0.0.0.0")
+        ip = request.headers.get("x-forwarded-for", request.META.get("REMOTE_ADDR", "0.0.0.0")).split(",")[0]
         user_agent = request.headers.get("user-agent", "???")[:255]
         qs = UserIP.objects.filter(user=request.user, ip=ip)
         if qs.exists():

src/ractf/management/commands/group_ips.py

@@ -0,0 +1,51 @@
+from json import dumps
+
+from django.core.management import BaseCommand
+
+from member.models import UserIP
+
+
+class Command(BaseCommand):
+    help = "Group users by source IP address to try and spot cheating."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--multiple",
+            action="store_true",
+            help="Show only IP addresses with multiple users",
+        )
+
+        parser.add_argument(
+            "--json",
+            action="store_true",
+            help="Output JSON",
+        )
+
+    def handle(self, *args, **options) -> None:
+        self.stderr.write(
+            self.style.WARNING("Due to use of CGNAT, source IP addresses may be unreliable. Proceed with caution.")
+        )
+
+        ips = UserIP.objects.all()
+
+        grouped = {}
+
+        for ip in ips:
+            if ip.ip not in grouped:
+                grouped[ip.ip] = []
+            if ip.user.username not in grouped[ip.ip]:
+                grouped[ip.ip].append(ip.user.username)
+
+        if options["multiple"]:
+            multiple_grouped = {}
+
+            for group in grouped.items():
+                if len(group[1]) > 1:
+                    multiple_grouped |= {group[0]: group[1]}
+
+            grouped = multiple_grouped
+
+        if options["json"]:
+            self.stdout.write(dumps(grouped))
+        else:
+            self.stdout.write(str(grouped))

src/ractf/tests.py

@@ -1 +1,45 @@
-# Create your tests here.
+from io import StringIO
+
+from django.contrib.auth import get_user_model
+from django.core.management import call_command
+from django.test import TestCase
+
+from member.models import UserIP
+
+
+class GroupIpsTest(TestCase):
+    def setUp(self):
+        one = get_user_model()(username="one", email="one@one.one")
+        one.save()
+
+        two = get_user_model()(username="two", email="two@two.two")
+        two.save()
+
+        three = get_user_model()(username="three", email="three@three.three")
+        three.save()
+
+        UserIP.objects.create(user=one, ip="1.1.1.1", user_agent="Django Tests")
+        UserIP.objects.create(user=two, ip="1.1.1.1", user_agent="Django Tests")
+        UserIP.objects.create(user=three, ip="2.2.2.2", user_agent="Django Tests")
+
+    def test_group_ips(self):
+        out = StringIO()
+        call_command("group_ips", stdout=out)
+        self.assertIn("1.1.1.1", out.getvalue())
+
+    def test_group_ips_multiple(self):
+        out = StringIO()
+        call_command("group_ips", "--multiple", stdout=out)
+        self.assertIn("1.1.1.1", out.getvalue())
+        self.assertNotIn("2.2.2.2", out.getvalue())
+
+    def test_group_ips_json(self):
+        out = StringIO()
+        call_command("group_ips", "--json", stdout=out)
+        self.assertIn("1.1.1.1", out.getvalue())
+
+    def test_group_ips_multiple_json(self):
+        out = StringIO()
+        call_command("group_ips", "--json", "--multiple", stdout=out)
+        self.assertIn("1.1.1.1", out.getvalue())
+        self.assertNotIn("2.2.2.2", out.getvalue())